Handle empty state case in global state manager

airbytehq · Jun 14, 2022 · 649effc · 649effc
1 parent e1ae1f7
commit 649effc
Showing 1 changed file with 10 additions and 2 deletions.
diff --git a/...b/src/main/java/io/airbyte/integrations/source/relationaldb/state/GlobalStateManager.java b/...b/src/main/java/io/airbyte/integrations/source/relationaldb/state/GlobalStateManager.java
@@ -20,6 +20,7 @@
 import io.airbyte.protocol.models.ConfiguredAirbyteCatalog;
 import io.airbyte.protocol.models.StreamDescriptor;
 import java.util.Collection;
+import java.util.List;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
@@ -110,12 +111,19 @@ private static Supplier<Collection<AirbyteStreamState>> getStreamsSupplier(final
      * we can look for streams in the "global" field of the message. Otherwise, the message is still
      * storing state in the legacy "data" field.
      */
-    return () -> airbyteStateMessage.getStateType() == AirbyteStateType.GLOBAL ? airbyteStateMessage.getGlobal().getStreamStates()
-        : Jsons.object(airbyteStateMessage.getData(), DbState.class).getStreams().stream()
+    return () -> {
+      if (airbyteStateMessage.getStateType() == AirbyteStateType.GLOBAL) {
+        return airbyteStateMessage.getGlobal().getStreamStates();
+      } else if (airbyteStateMessage.getData() != null) {
+        return Jsons.object(airbyteStateMessage.getData(), DbState.class).getStreams().stream()
             .map(s -> new AirbyteStreamState().withStreamState(Jsons.jsonNode(s))
                 .withStreamDescriptor(new StreamDescriptor().withNamespace(s.getStreamNamespace()).withName(s.getStreamName())))
             .collect(
                 Collectors.toList());
+      } else {
+        return List.of();
+      }
+    };
   }
 
 }