Merge remote-tracking branch 'origin/snoflake-2.1.2-upgrade' into sno…

…flake-2.1.2-upgrade # Conflicts: # pom.xml # src/main/java/com/snowflake/kafka/connector/internal/streaming/SchematizationUtils.java # src/main/java/com/snowflake/kafka/connector/records/RecordService.java
streamkap-com · Feb 1, 2024 · 75e17d0 · 75e17d0
2 parents 50c1c2e + e4b9b0d
commit 75e17d0
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 4 deletions.
diff --git a/src/main/java/com/snowflake/kafka/connector/records/RecordService.java b/src/main/java/com/snowflake/kafka/connector/records/RecordService.java
@@ -308,10 +308,7 @@ private Map<String, Object> getMapFromJsonNodeForStreamingIngest(JsonNode node)
       String columnName = columnNames.next();
       JsonNode columnNode = node.get(columnName);
       Object columnValue;
-      if (columnNode.isBinary()) {
-        byte[] binaryValue = Base64.getDecoder().decode(columnNode.asText());
-        columnValue = HexFormat.of().formatHex(binaryValue);
-      } else if (columnNode.isTextual()) {
+      if (columnNode.isTextual()) {
         columnValue = columnNode.textValue();
       } else if (columnNode.isBinary()) {
         byte[] binaryValue = Base64.getDecoder().decode(columnNode.asText());

diff --git a/src/test/java/com/snowflake/kafka/connector/UtilsTest.java b/src/test/java/com/snowflake/kafka/connector/UtilsTest.java
@@ -141,6 +141,14 @@ public void testTableNameRegex() {
     assert Utils.tableName(topic, topic2table).equals("bird_" + Math.abs(topic.hashCode()));
   }
 
+  @Test
+  public void test_generateValidName() {
+    Map<String, String> topic2table = Utils.parseTopicToTableMap("ab@cd:abcd, REGEX_MATCHER>^[^.]\\w+.\\w+.(.*):$1, 1234:_1234");
+
+    assert Utils.tableName("postgres.public.testtbl", topic2table).equals("testtbl");
+  }
+
+
   @Test
   public void testTableFullName() {
     assert Utils.isValidSnowflakeTableName("_1342dfsaf$");