move state manager inside incremental block

airbytehq · Dec 3, 2020 · 775e2ce · 775e2ce
1 parent f9eceb8
commit 775e2ce
Showing 1 changed file with 1 addition and 1 deletion.
diff --git a/...ors/source-jdbc/src/main/java/io/airbyte/integrations/source/jdbc/AbstractJdbcSource.java b/...ors/source-jdbc/src/main/java/io/airbyte/integrations/source/jdbc/AbstractJdbcSource.java
@@ -153,7 +153,6 @@ private List<Table<?>> discoverInternal(final Database database) throws Exceptio
 
   @Override
   public Stream<AirbyteMessage> read(JsonNode config, ConfiguredAirbyteCatalog catalog, JsonNode state) throws Exception {
-    final JdbcStateManager stateManager = new JdbcStateManager(Jsons.object(state, JdbcState.class));
     final Instant now = Instant.now();
 
     final Database database = createDatabase(config);
@@ -185,6 +184,7 @@ public Stream<AirbyteMessage> read(JsonNode config, ConfiguredAirbyteCatalog cat
 
       final Stream<AirbyteMessage> stream;
       if (airbyteStream.getSyncMode() == SyncMode.INCREMENTAL) {
+        final JdbcStateManager stateManager = new JdbcStateManager(Jsons.object(state, JdbcState.class));
         final String cursorField = IncrementalUtils.getCursorField(airbyteStream);
         final JsonSchemaPrimitive cursorType = IncrementalUtils.getCursorType(airbyteStream, cursorField);
         final Optional<String> initialCursorOptional = stateManager.getOriginalCursor(streamName);