remove unneeded observe() and update comment

Author: brianjlai

airbyte_cdk/sources/concurrent_source/concurrent_read_processor.py

@@ -154,7 +154,6 @@ def on_record(self, record: Record) -> Iterable[AirbyteMessage]:
                     stream.as_airbyte_stream(), AirbyteStreamStatus.RUNNING
                 )
             self._record_counter[stream.name] += 1
-            stream.cursor.observe(record)
         yield message
         yield from self._message_repository.consume_queue()
 

airbyte_cdk/sources/streams/concurrent/cursor.py

@@ -230,13 +230,10 @@ def _get_concurrent_state(
 
     def observe(self, record: Record) -> None:
         # Because observe writes to the most_recent_cursor_value_per_partition mapping,
-        # it is not thread-safe. However, this shouldn't lead to concurrency issues
-        # because observe() is only invoked in two ways both of which aren't conflicting:
-        # - ConcurrentReadProcessor.on_record(): Since records are observed on the main
-        # thread and so there aren't concurrent operations. Partitions are also split by key.
-        # - PartitionReader.process_partition(): Because the map is broken down according to
-        # partition, concurrent threads processing only read/write from different keys which
-        # avoids any conflicts.
+        # it is not thread-safe. However, this shouldn't lead to concurrency issues because
+        # observe() is only invoked by PartitionReader.process_partition(). Since the map is
+        # broken down according to partition, concurrent threads processing only read/write
+        # from different keys which avoids any conflicts.
         #
         # If we were to add thread safety, we should implement a lock per-partition
         # which is instantiated during stream_slices()