HGI-6439: updates lastest_state

target_api/target.py

@@ -78,5 +78,21 @@ def _process_record_message(self, message_dict: dict) -> None:
             sink.process_record(transformed_record, context)
             sink._after_process_record(context)
 
+            if sink.is_full:
+                self.logger.info(
+                    f"Target sink for '{sink.stream_name}' is full. Draining..."
+                )
+                self.drain_one(sink)
+
+            if not self._latest_state:
+                # If "self._latest_state" is empty, save the value of "sink.latest_state"
+                self._latest_state = sink.latest_state
+            else:
+                # If "self._latest_state" is not empty, update all its fields with the
+                # fields from "sink.latest_state" (if they exist)
+                for key in self._latest_state.keys():
+                    sink_latest_state = sink.latest_state or dict()
+                    self._latest_state[key].update(sink_latest_state.get(key) or dict())
+
 if __name__ == "__main__":
     TargetApi.cli()