[source-postgres] fix on final state (#44119)

Author: xiaohansong

airbyte-integrations/connectors/source-postgres/integration_tests/seed/hook.py

@@ -10,6 +10,7 @@
 from typing import List, Tuple
 
 import psycopg2
+import pytz
 from psycopg2 import extensions, sql
 
 catalog_write_file = "/connector/integration_tests/temp/configured_catalog_copy.json"
@@ -24,6 +25,8 @@
 secret_config_cdc_file = '/connector/secrets/config_cdc.json'
 secret_active_config_cdc_file = '/connector/integration_tests/temp/config_cdc_active.json'
 
+la_timezone = pytz.timezone('America/Los_Angeles')
+
 def connect_to_db() -> extensions.connection:
     with open(secret_config_file) as f:
         secret = json.load(f)
@@ -124,7 +127,7 @@ def create_table(conn: extensions.connection, schema_name: str, table_name: str)
         cursor.close()
 
 def generate_schema_date_with_suffix() -> str:
-    current_date = datetime.datetime.now().strftime("%Y%m%d")
+    current_date = datetime.datetime.now(la_timezone).strftime("%Y%m%d")
     suffix = ''.join(random.choices(string.ascii_lowercase + string.digits, k=8))
     return f"{current_date}_{suffix}"
 
@@ -251,7 +254,7 @@ def delete_schemas_with_prefix(conn, date_prefix):
 
 def teardown() -> None:
     conn = connect_to_db()
-    today = datetime.datetime.now()
+    today = datetime.datetime.now(la_timezone)
     yesterday = today - timedelta(days=1)
     formatted_yesterday = yesterday.strftime('%Y%m%d')
     delete_schemas_with_prefix(conn, formatted_yesterday)

airbyte-integrations/connectors/source-postgres/integration_tests/seed/requirement.txt

@@ -1 +1,2 @@
 psycopg2==2.9.9
+pytz

airbyte-integrations/connectors/source-postgres/metadata.yaml

@@ -9,7 +9,7 @@ data:
   connectorSubtype: database
   connectorType: source
   definitionId: decd338e-5647-4c0b-adf4-da0e75f5a750
-  dockerImageTag: 3.6.15
+  dockerImageTag: 3.6.16
   dockerRepository: airbyte/source-postgres
   documentationUrl: https://docs.airbyte.com/integrations/sources/postgres
   githubIssueLabel: source-postgres

airbyte-integrations/connectors/source-postgres/src/main/java/io/airbyte/integrations/source/postgres/ctid/CtidGlobalStateManager.java

@@ -106,7 +106,6 @@ public AirbyteStateMessage createCtidStateMessage(final AirbyteStreamNameNamespa
     streamsThatHaveCompletedSnapshot.forEach(stream -> {
       final DbStreamState state = getFinalState(stream);
       streamStates.add(getAirbyteStreamState(stream, Jsons.jsonNode(state)));
-
     });
 
     resumableFullRefreshStreams.forEach(stream -> {
@@ -159,14 +158,20 @@ public AirbyteStateMessage createFinalStateMessage(final AirbyteStreamNameNamesp
     if (isIncrementalStream(pair)) {
       streamsThatHaveCompletedSnapshot.add(pair);
     }
+
+    if (resumableFullRefreshStreams.contains(pair)) {
+      final CtidStatus ctidStatusForFullRefreshStream = generateCtidStatusForState(pair);
+      pairToCtidStatus.put(pair, ctidStatusForFullRefreshStream);
+    }
+
     final List<AirbyteStreamState> streamStates = new ArrayList<>();
     streamsThatHaveCompletedSnapshot.forEach(stream -> {
       final DbStreamState state = getFinalState(stream);
       streamStates.add(getAirbyteStreamState(stream, Jsons.jsonNode(state)));
     });
 
     resumableFullRefreshStreams.forEach(stream -> {
-      final CtidStatus ctidStatusForFullRefreshStream = generateCtidStatusForState(pair);
+      final CtidStatus ctidStatusForFullRefreshStream = generateCtidStatusForState(stream);
       streamStates.add(getAirbyteStreamState(stream, Jsons.jsonNode(ctidStatusForFullRefreshStream)));
     });