Move acquiring the semaphore

Author: tolik0

airbyte_cdk/sources/declarative/incremental/concurrent_partition_cursor.py

@@ -146,6 +146,7 @@ def close_partition(self, partition: Partition) -> None:
 
         partition_key = self._to_partition_key(stream_slice.partition)
         with self._lock:
+            self._semaphore_per_partition[partition_key].acquire()
             if not self._use_global_cursor:
                 self._cursor_per_partition[partition_key].close_partition(partition=partition)
                 cursor = self._cursor_per_partition[partition_key]
@@ -156,8 +157,6 @@ def close_partition(self, partition: Partition) -> None:
                     self._update_global_cursor(cursor.state[self.cursor_field.cursor_field_key])
                 self._emit_state_message()
 
-            self._semaphore_per_partition[partition_key].acquire()
-
     def ensure_at_least_one_state_emitted(self) -> None:
         """
         The platform expect to have at least one state message on successful syncs. Hence, whatever happens, we expect this method to be