🐛 Source Relational DB : remove unicode nulls from cursors (#13854)

* Make sure that cursor value doesn't contain unicode null. * Test that cursors doesn't contain unicode null * incr version
airbytehq · Jun 17, 2022 · 6ba511c · 6ba511c
1 parent 074adef
commit 6ba511c
Show file tree

Hide file tree

Showing 3 changed files with 30 additions and 2 deletions.
diff --git a/airbyte-integrations/connectors/source-relational-db/Dockerfile b/airbyte-integrations/connectors/source-relational-db/Dockerfile
@@ -16,5 +16,5 @@ ENV APPLICATION source-relational-db
 
 COPY --from=build /airbyte /airbyte
 
-LABEL io.airbyte.version=0.3.0
+LABEL io.airbyte.version=0.3.1
 LABEL io.airbyte.name=airbyte/source-relational-db
diff --git a/...db/src/main/java/io/airbyte/integrations/source/relationaldb/StateDecoratingIterator.java b/...db/src/main/java/io/airbyte/integrations/source/relationaldb/StateDecoratingIterator.java
@@ -43,12 +43,17 @@ public StateDecoratingIterator(final Iterator<AirbyteMessage> messageIterator,
     stateManager.setIsCdc(false);
   }
 
+  private String getCursorCandidate(final AirbyteMessage message) {
+    String cursorCandidate = message.getRecord().getData().get(cursorField).asText();
+    return (cursorCandidate != null ? cursorCandidate.replaceAll("\u0000", "") : null);
+  }
+
   @Override
   protected AirbyteMessage computeNext() {
     if (messageIterator.hasNext()) {
       final AirbyteMessage message = messageIterator.next();
       if (message.getRecord().getData().hasNonNull(cursorField)) {
-        final String cursorCandidate = message.getRecord().getData().get(cursorField).asText();
+        final String cursorCandidate = getCursorCandidate(message);
         if (IncrementalUtils.compareCursors(maxCursor, cursorCandidate, cursorType) < 0) {
           maxCursor = cursorCandidate;
         }

diff --git a/...rc/test/java/io/airbyte/integrations/source/relationaldb/StateDecoratingIteratorTest.java b/...rc/test/java/io/airbyte/integrations/source/relationaldb/StateDecoratingIteratorTest.java
@@ -41,6 +41,11 @@ class StateDecoratingIteratorTest {
       .withRecord(new AirbyteRecordMessage()
           .withData(Jsons.jsonNode(ImmutableMap.of(UUID_FIELD_NAME, "def"))));
 
+  private static final AirbyteMessage RECORD_MESSAGE3 = new AirbyteMessage()
+      .withType(Type.RECORD)
+      .withRecord(new AirbyteRecordMessage()
+          .withData(Jsons.jsonNode(ImmutableMap.of(UUID_FIELD_NAME, "abc\u0000"))));
+
   private static Iterator<AirbyteMessage> messageIterator;
   private StateManager stateManager;
   private AirbyteStateMessage stateMessage;
@@ -130,4 +135,22 @@ void testEmptyStream() {
     assertFalse(iterator.hasNext());
   }
 
+  @Test
+  void testUnicodeNull() {
+    messageIterator = MoreIterators.of(RECORD_MESSAGE3);
+    when(stateManager.updateAndEmit(NAME_NAMESPACE_PAIR, "abc")).thenReturn(stateMessage);
+
+    final StateDecoratingIterator iterator = new StateDecoratingIterator(
+        messageIterator,
+        stateManager,
+        NAME_NAMESPACE_PAIR,
+        UUID_FIELD_NAME,
+        null,
+        JsonSchemaPrimitive.STRING);
+
+    assertEquals(RECORD_MESSAGE3, iterator.next());
+    assertEquals(stateMessage, iterator.next().getState());
+    assertFalse(iterator.hasNext());
+  }
+
 }