Fix Avro ts mapping in case of avro-confluent format in AvroTableSchemaFactory (#1661)

Author: ferenc-csaky

sqrl-planner/src/main/java/com/datasqrl/io/schema/avro/AvroTableSchemaFactory.java

@@ -64,16 +64,33 @@ public Set<String> getExtensions() {
   }
 
   /**
-   * Extracts the legacy timestamp mapping setting from table properties, checking both direct and
-   * value-prefixed property keys. Returns false if the keys are null or undefined.
+   * Extracts the legacy timestamp mapping setting from table properties.
+   *
+   * <p>Checks for the property in the following order:
+   *
+   * <ol>
+   *   <li>Direct property key
+   *   <li>Value-prefixed property key
+   *   <li>If not found and any property key contains "avro-confluent", returns true
+   *   <li>Otherwise returns false
+   * </ol>
+   *
+   * @param tableProps the table properties to extract the setting from
+   * @return true if legacy timestamp mapping should be used, false otherwise
    */
-  private boolean getLegacyTimestampMapping(Map<String, String> tableProps) {
+  boolean getLegacyTimestampMapping(Map<String, String> tableProps) {
     var legacyTimestampMappingStr = tableProps.get(LEGACY_TIMESTAMP_MAPPING_KEY);
 
     if (legacyTimestampMappingStr == null) {
       legacyTimestampMappingStr = tableProps.get("value." + LEGACY_TIMESTAMP_MAPPING_KEY);
     }
 
+    // avro-confluent uses legacy no matter what
+    if (legacyTimestampMappingStr == null
+        && tableProps.keySet().stream().anyMatch(key -> key.contains("avro-confluent"))) {
+      return true;
+    }
+
     return Boolean.parseBoolean(legacyTimestampMappingStr);
   }
 }

sqrl-planner/src/test/java/com/datasqrl/io/schema/avro/AvroTableSchemaFactoryTest.java

@@ -0,0 +1,96 @@
+/*
+ * Copyright © 2021 DataSQRL ([email protected])
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datasqrl.io.schema.avro;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.HashMap;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+class AvroTableSchemaFactoryTest {
+
+  private AvroTableSchemaFactory factory;
+
+  @BeforeEach
+  void setUp() {
+    factory = new AvroTableSchemaFactory();
+  }
+
+  @Test
+  void givenDirectPropertyKeySetToTrue_whenGetLegacyTimestampMapping_thenReturnsTrue() {
+    // Given
+    var tableProps = new HashMap<String, String>();
+    tableProps.put("avro.timestamp_mapping.legacy", "true");
+
+    // When
+    var result = factory.getLegacyTimestampMapping(tableProps);
+
+    // Then
+    assertThat(result).isTrue();
+  }
+
+  @Test
+  void givenDirectPropertyKeySetToFalse_whenGetLegacyTimestampMapping_thenReturnsFalse() {
+    // Given
+    var tableProps = new HashMap<String, String>();
+    tableProps.put("avro.timestamp_mapping.legacy", "false");
+
+    // When
+    var result = factory.getLegacyTimestampMapping(tableProps);
+
+    // Then
+    assertThat(result).isFalse();
+  }
+
+  @Test
+  void givenValuePrefixedPropertyKeySetToTrue_whenGetLegacyTimestampMapping_thenReturnsTrue() {
+    // Given
+    var tableProps = new HashMap<String, String>();
+    tableProps.put("value.avro.timestamp_mapping.legacy", "true");
+
+    // When
+    var result = factory.getLegacyTimestampMapping(tableProps);
+
+    // Then
+    assertThat(result).isTrue();
+  }
+
+  @Test
+  void givenNoPropertiesButAvroConfluentKey_whenGetLegacyTimestampMapping_thenReturnsTrue() {
+    // Given
+    var tableProps = new HashMap<String, String>();
+    tableProps.put("avro-confluent.url", "registry");
+
+    // When
+    var result = factory.getLegacyTimestampMapping(tableProps);
+
+    // Then
+    assertThat(result).isTrue();
+  }
+
+  @Test
+  void givenEmptyTableProps_whenGetLegacyTimestampMapping_thenReturnsFalse() {
+    // Given
+    var tableProps = new HashMap<String, String>();
+
+    // When
+    var result = factory.getLegacyTimestampMapping(tableProps);
+
+    // Then
+    assertThat(result).isFalse();
+  }
+}