Merge branch 'tolik0/refactor-concurrent-global-cursor' into tolik0/concurrent-perpartition-add-parent-state-updates

Author: tolik0

airbyte_cdk/sources/declarative/incremental/concurrent_partition_cursor.py

@@ -150,21 +150,20 @@ 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]
                 if (
-                        partition_key in self._finished_partitions
-                        and self._semaphore_per_partition[partition_key]._value == 0
+                    partition_key in self._finished_partitions
+                    and self._semaphore_per_partition[partition_key]._value == 0
                 ):
                     self._update_global_cursor(cursor.state[self.cursor_field.cursor_field_key])
 
             self._check_and_update_parent_state()
 
             self._emit_state_message()
 
-            self._semaphore_per_partition[partition_key].acquire()
-
     def _check_and_update_parent_state(self) -> None:
         """
         If all slices for the earliest partitions are closed, pop them from the left