Marking WorkItems DONE before releasing the lock.

Without this, there is the slight chance of an extra progress report being sent
resulting in unnecessary lease expiration errors.

----Release Notes----
[]
-------------
Created by MOE: https://github.com/google/moe
MOE_MIGRATED_REVID=117271848

Author: chamikaramj

google/cloud/dataflow/worker/batchworker.py

@@ -407,12 +407,12 @@ def do_work(self, work_item):
       with work_item.lock:
         self.report_completion_status(work_item,
                                       exception_details=exception_details)
+        work_item.done = True
     else:
       self.report_progress = False
       with work_item.lock:
         self.report_completion_status(work_item)
-    with work_item.lock:
-      work_item.done = True
+        work_item.done = True
 
   def status_server(self):
     """Executes the serving loop for the status server."""