CONN-10463 Revert role validation + telemetry improvements + Java 11 fixes (#1133)

sfc-gh-mbobowski · web-flow · commit a34681e5685e · 2025-07-07T13:40:17.000+02:00
diff --git a/src/main/java/com/snowflake/kafka/connector/SnowflakeSinkConnectorConfig.java b/src/main/java/com/snowflake/kafka/connector/SnowflakeSinkConnectorConfig.java
@@ -19,7 +19,6 @@
 import static com.snowflake.kafka.connector.Utils.isSnowpipeStreamingIngestion;
 
 import com.google.common.base.Strings;
-import com.google.common.collect.ImmutableSet;
 import com.snowflake.kafka.connector.internal.KCLogger;
 import com.snowflake.kafka.connector.internal.streaming.IngestionMethodConfig;
 import java.util.Arrays;
@@ -251,7 +250,7 @@ public class SnowflakeSinkConnectorConfig {
       "value.converter.schema.registry.url";
 
   public static final Set<String> CUSTOM_SNOWFLAKE_CONVERTERS =
-      ImmutableSet.of(
+      Set.of(
           "com.snowflake.kafka.connector.records.SnowflakeJsonConverter",
           "com.snowflake.kafka.connector.records.SnowflakeAvroConverterWithoutSchemaRegistry",
           "com.snowflake.kafka.connector.records.SnowflakeAvroConverter");
diff --git a/src/main/java/com/snowflake/kafka/connector/internal/streaming/DefaultStreamingConfigValidator.java b/src/main/java/com/snowflake/kafka/connector/internal/streaming/DefaultStreamingConfigValidator.java
@@ -101,35 +101,13 @@ public ImmutableMap<String, String> validate(Map<String, String> inputConfig) {
   }
 
   private static Optional<Map.Entry<String, String>> validateRole(Map<String, String> inputConfig) {
-    if (Utils.isSnowpipeStreamingV2Enabled(inputConfig)) {
-      return validateRoleForSSv2(inputConfig);
-    } else {
-      return validateRoleForSSv1(inputConfig);
-    }
-  }
-
-  private static Optional<Map.Entry<String, String>> validateRoleForSSv1(
-      Map<String, String> inputConfig) {
     if (!inputConfig.containsKey(Utils.SF_ROLE)
         || Strings.isNullOrEmpty(inputConfig.get(Utils.SF_ROLE))) {
-      String roleMissingForSSv1 =
+      String missingRole =
           String.format(
               "Config:%s should be present if ingestionMethod is:%s",
               Utils.SF_ROLE, inputConfig.get(INGESTION_METHOD_OPT));
-      return Optional.of(Map.entry(Utils.SF_ROLE, roleMissingForSSv1));
-    }
-    return Optional.empty();
-  }
-
-  private static Optional<Map.Entry<String, String>> validateRoleForSSv2(
-      Map<String, String> inputConfig) {
-    if (inputConfig.containsKey(Utils.SF_ROLE)) {
-      String rolePresentForSSv2 =
-          String.format(
-              "The default role is used when '%s' is enabled. Delete '%s' parameter from the"
-                  + " connector config and make sure that default role is set properly.",
-              SnowflakeSinkConnectorConfig.SNOWPIPE_STREAMING_V2_ENABLED, Utils.SF_ROLE);
-      return Optional.of(Map.entry(Utils.SF_ROLE, rolePresentForSSv2));
+      return Optional.of(Map.entry(Utils.SF_ROLE, missingRole));
     }
     return Optional.empty();
   }
diff --git a/src/main/java/com/snowflake/kafka/connector/internal/streaming/schemaevolution/InsertErrorMapper.java b/src/main/java/com/snowflake/kafka/connector/internal/streaming/schemaevolution/InsertErrorMapper.java
@@ -1,6 +1,5 @@
 package com.snowflake.kafka.connector.internal.streaming.schemaevolution;
 
-import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Iterables;
 import com.google.common.collect.Lists;
 import java.util.List;
@@ -21,10 +20,10 @@ public SchemaEvolutionTargetItems mapToSchemaEvolutionItems(
         extraColNames);
   }
 
-  private List<String> joinNullableLists(List<String> list1, List<String> list2) {
+  private static List<String> joinNullableLists(List<String> list1, List<String> list2) {
     return Lists.newArrayList(
         Iterables.concat(
-            Optional.ofNullable(list1).orElse(ImmutableList.of()),
-            Optional.ofNullable(list2).orElse(ImmutableList.of())));
+            Optional.ofNullable(list1).orElse(List.of()),
+            Optional.ofNullable(list2).orElse(List.of())));
   }
 }
diff --git a/src/main/java/com/snowflake/kafka/connector/internal/streaming/schemaevolution/iceberg/IcebergTableSchemaResolver.java b/src/main/java/com/snowflake/kafka/connector/internal/streaming/schemaevolution/iceberg/IcebergTableSchemaResolver.java
@@ -1,7 +1,6 @@
 package com.snowflake.kafka.connector.internal.streaming.schemaevolution.iceberg;
 
 import com.fasterxml.jackson.databind.JsonNode;
-import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Streams;
 import com.snowflake.kafka.connector.internal.SnowflakeErrors;
 import com.snowflake.kafka.connector.internal.streaming.common.ColumnProperties;
@@ -43,7 +42,7 @@ public List<IcebergColumnTree> resolveIcebergSchemaFromChannel(
   public List<IcebergColumnTree> resolveIcebergSchemaFromRecord(
       SinkRecord record, Set<String> columnsToEvolve) {
     if (columnsToEvolve == null || columnsToEvolve.isEmpty()) {
-      return ImmutableList.of();
+      return List.of();
     }
     if (hasSchema(record)) {
       LOGGER.debug(
diff --git a/src/main/java/com/snowflake/kafka/connector/internal/telemetry/SnowflakeTelemetryService.java b/src/main/java/com/snowflake/kafka/connector/internal/telemetry/SnowflakeTelemetryService.java
@@ -263,6 +263,17 @@ protected void addUserConnectorPropertiesToDataNode(
         userProvidedConfig.getOrDefault(
             ENABLE_SCHEMATIZATION_CONFIG, ENABLE_SCHEMATIZATION_DEFAULT));
 
+    dataObjectNode.put(
+        SNOWPIPE_STREAMING_V2_ENABLED,
+        userProvidedConfig.getOrDefault(
+            SNOWPIPE_STREAMING_V2_ENABLED,
+            String.valueOf(SNOWPIPE_STREAMING_V2_ENABLED_DEFAULT_VALUE)));
+
+    dataObjectNode.put(
+        ICEBERG_ENABLED,
+        userProvidedConfig.getOrDefault(
+            ICEBERG_ENABLED, String.valueOf(ICEBERG_ENABLED_DEFAULT_VALUE)));
+
     // Record whether streaming client optimization is enabled
     dataObjectNode.put(
         ENABLE_STREAMING_CLIENT_OPTIMIZATION_CONFIG,
diff --git a/src/test/java/com/snowflake/kafka/connector/ConnectorConfigValidatorTest.java b/src/test/java/com/snowflake/kafka/connector/ConnectorConfigValidatorTest.java
@@ -924,19 +924,18 @@ public void shouldValidateSSv2Config() {
     Map<String, String> config =
         SnowflakeSinkConnectorConfigBuilder.streamingConfig()
             .withSnowpipeStreamingV2Enabled()
-            .withoutRole()
             .build();
 
     assertThatCode(() -> connectorConfigValidator.validateConfig(config))
         .doesNotThrowAnyException();
   }
 
   @Test
-  public void shouldThrowExceptionWhenRoleDefinedForSSv2() {
+  public void shouldThrowExceptionWhenRoleNotDefinedForSSv2() {
     Map<String, String> config =
         SnowflakeSinkConnectorConfigBuilder.streamingConfig()
             .withSnowpipeStreamingV2Enabled()
-            .withRole("someRole")
+            .withoutRole()
             .build();
 
     assertThatThrownBy(() -> connectorConfigValidator.validateConfig(config))
diff --git a/src/test/java/com/snowflake/kafka/connector/config/IcebergConfigValidationTest.java b/src/test/java/com/snowflake/kafka/connector/config/IcebergConfigValidationTest.java
@@ -3,7 +3,6 @@
 import static com.snowflake.kafka.connector.SnowflakeSinkConnectorConfig.INGESTION_METHOD_OPT;
 import static com.snowflake.kafka.connector.SnowflakeSinkConnectorConfig.SNOWPIPE_STREAMING_V2_ENABLED;
 
-import com.google.common.collect.ImmutableMap;
 import com.snowflake.kafka.connector.internal.streaming.IngestionMethodConfig;
 import com.snowflake.kafka.connector.internal.streaming.StreamingConfigValidator;
 import java.util.Map;
@@ -21,7 +20,7 @@ public class IcebergConfigValidationTest {
   @MethodSource("validConfigs")
   public void shouldValidateCorrectConfig(Map<String, String> config) {
     // when
-    ImmutableMap<String, String> invalidParameters = validator.validate(config);
+    Map<String, String> invalidParameters = validator.validate(config);
 
     // then
     Assertions.assertTrue(invalidParameters.isEmpty());
@@ -31,7 +30,7 @@ public void shouldValidateCorrectConfig(Map<String, String> config) {
   @MethodSource("invalidConfigs")
   public void shouldReturnErrorOnInvalidConfig(Map<String, String> config, String errorKey) {
     // when
-    ImmutableMap<String, String> invalidParameters = validator.validate(config);
+    Map<String, String> invalidParameters = validator.validate(config);
 
     // then
     Assertions.assertTrue(invalidParameters.containsKey(errorKey));
diff --git a/src/test/java/com/snowflake/kafka/connector/internal/OAuthURLTest.java b/src/test/java/com/snowflake/kafka/connector/internal/OAuthURLTest.java
@@ -3,7 +3,6 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 
-import com.google.common.collect.ImmutableList;
 import java.util.List;
 import java.util.stream.Stream;
 import org.junit.jupiter.api.Test;
@@ -17,11 +16,10 @@ static Stream<Arguments> correctUrls() {
     return Stream.of(
         Arguments.of(
             "https://localhost:8085/push/token",
-            ImmutableList.of("https", "localhost:8085", "/push/token", true)),
-        Arguments.of("localhost:8085", ImmutableList.of("https", "localhost:8085", "", true)),
-        Arguments.of(
-            "http://localhost:8085", ImmutableList.of("http", "localhost:8085", "", false)),
-        Arguments.of("localhost", ImmutableList.of("https", "localhost:443", "", true)));
+            List.of("https", "localhost:8085", "/push/token", true)),
+        Arguments.of("localhost:8085", List.of("https", "localhost:8085", "", true)),
+        Arguments.of("http://localhost:8085", List.of("http", "localhost:8085", "", false)),
+        Arguments.of("localhost", List.of("https", "localhost:443", "", true)));
   }
 
   @ParameterizedTest(name = "url: {0}, parsed: {1}")
diff --git a/src/test/java/com/snowflake/kafka/connector/internal/TestUtils.java b/src/test/java/com/snowflake/kafka/connector/internal/TestUtils.java
@@ -31,8 +31,6 @@
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
-import com.google.common.collect.ImmutableSet;
-import com.google.common.collect.Sets;
 import com.snowflake.kafka.connector.SnowflakeSinkConnectorConfig;
 import com.snowflake.kafka.connector.Utils;
 import com.snowflake.kafka.connector.config.SnowflakeSinkConnectorConfigBuilder;
@@ -61,9 +59,6 @@
 import java.util.function.Function;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
-import java.util.stream.Collectors;
-import java.util.stream.IntStream;
-import java.util.stream.Stream;
 import net.snowflake.client.jdbc.internal.apache.http.HttpHeaders;
 import net.snowflake.client.jdbc.internal.apache.http.client.methods.CloseableHttpResponse;
 import net.snowflake.client.jdbc.internal.apache.http.client.methods.HttpPost;
@@ -81,7 +76,6 @@
 import org.apache.kafka.connect.data.Struct;
 import org.apache.kafka.connect.json.JsonConverter;
 import org.apache.kafka.connect.sink.SinkRecord;
-import org.junit.jupiter.params.provider.Arguments;
 
 public class TestUtils {
   // test profile properties
@@ -1029,14 +1023,4 @@ private static StringEntity buildStringEntity(String payload) {
       throw new RuntimeException(e);
     }
   }
-
-  public static Stream<Arguments> nBooleanProduct(int n) {
-    return Sets.cartesianProduct(
-            IntStream.range(0, n)
-                .mapToObj(i -> ImmutableSet.of(false, true))
-                .collect(Collectors.toList()))
-        .stream()
-        .map(List::toArray)
-        .map(Arguments::of);
-  }
 }
diff --git a/src/test/java/com/snowflake/kafka/connector/internal/TombstoneRecordIngestionIT.java b/src/test/java/com/snowflake/kafka/connector/internal/TombstoneRecordIngestionIT.java
@@ -3,8 +3,6 @@
 import static com.snowflake.kafka.connector.ConnectorConfigValidatorTest.COMMUNITY_CONVERTER_SUBSET;
 import static com.snowflake.kafka.connector.ConnectorConfigValidatorTest.CUSTOM_SNOWFLAKE_CONVERTERS;
 
-import com.google.common.collect.ImmutableSet;
-import com.google.common.collect.Sets;
 import com.snowflake.kafka.connector.SnowflakeSinkConnectorConfig;
 import com.snowflake.kafka.connector.Utils;
 import com.snowflake.kafka.connector.dlq.InMemoryKafkaRecordErrorReporter;
@@ -20,7 +18,6 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.stream.Stream;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.connect.data.Schema;
 import org.apache.kafka.connect.data.SchemaAndValue;
@@ -30,9 +27,7 @@
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.EnumSource;
-import org.junit.jupiter.params.provider.MethodSource;
 
 public class TombstoneRecordIngestionIT {
   private final int partition = 0;
@@ -56,16 +51,8 @@ public void afterEach() {
     TestUtils.dropTable(table);
   }
 
-  private static Stream<Arguments> behaviorAndSingleBufferParameters() {
-    return Sets.cartesianProduct(
-            ImmutableSet.copyOf(SnowflakeSinkConnectorConfig.BehaviorOnNullValues.values()))
-        .stream()
-        .map(List::toArray)
-        .map(Arguments::of);
-  }
-
   @ParameterizedTest(name = "behavior: {0}")
-  @MethodSource("behaviorAndSingleBufferParameters")
+  @EnumSource(SnowflakeSinkConnectorConfig.BehaviorOnNullValues.class)
   public void testStreamingTombstoneBehavior(
       SnowflakeSinkConnectorConfig.BehaviorOnNullValues behavior) throws Exception {
     // setup
@@ -93,7 +80,7 @@ public void testStreamingTombstoneBehavior(
   }
 
   @ParameterizedTest(name = "behavior: {0}")
-  @MethodSource("behaviorAndSingleBufferParameters")
+  @EnumSource(SnowflakeSinkConnectorConfig.BehaviorOnNullValues.class)
   public void testStreamingTombstoneBehaviorWithSchematization(
       SnowflakeSinkConnectorConfig.BehaviorOnNullValues behavior) throws Exception {
     // setup
diff --git a/src/test/java/com/snowflake/kafka/connector/records/AbstractMetaColumnTest.java b/src/test/java/com/snowflake/kafka/connector/records/AbstractMetaColumnTest.java
@@ -20,7 +20,6 @@
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.ImmutableSet;
 import com.snowflake.kafka.connector.builder.SinkRecordBuilder;
 import com.snowflake.kafka.connector.mock.MockSchemaRegistryClient;
 import java.io.IOException;
@@ -122,13 +121,13 @@ static Stream<Arguments> dataFor_disablingFields() {
     return Stream.of(
         arguments(
             ImmutableMap.of(SNOWFLAKE_METADATA_CREATETIME, "false"),
-            ImmutableSet.of(TOPIC, PARTITION, OFFSET)),
+            Set.of(TOPIC, PARTITION, OFFSET)),
         arguments(
             ImmutableMap.of(SNOWFLAKE_METADATA_TOPIC, "false"),
-            ImmutableSet.of(PARTITION, OFFSET, TimestampType.CREATE_TIME.name)),
+            Set.of(PARTITION, OFFSET, TimestampType.CREATE_TIME.name)),
         arguments(
             ImmutableMap.of(SNOWFLAKE_METADATA_OFFSET_AND_PARTITION, "false"),
-            ImmutableSet.of(TOPIC, TimestampType.CREATE_TIME.name)),
+            Set.of(TOPIC, TimestampType.CREATE_TIME.name)),
         arguments(
             ImmutableMap.of(
                 SNOWFLAKE_METADATA_CREATETIME,
diff --git a/src/test/java/com/snowflake/kafka/connector/records/IcebergTableStreamingRecordMapperTest.java b/src/test/java/com/snowflake/kafka/connector/records/IcebergTableStreamingRecordMapperTest.java
@@ -4,11 +4,11 @@
 import static org.assertj.core.api.Assertions.assertThat;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
-import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.snowflake.kafka.connector.Utils;
 import com.snowflake.kafka.connector.records.RecordService.SnowflakeTableRow;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.stream.Stream;
 import org.junit.jupiter.api.Test;
@@ -204,14 +204,14 @@ private static Stream<Arguments> prepareSchematizationData() throws JsonProcessi
                 "{\"key\": [" + primitiveJsonExample + ", " + primitiveJsonExample + "]}"),
             ImmutableMap.of(
                 "\"KEY\"",
-                ImmutableList.of(primitiveJsonAsMap, primitiveJsonAsMap),
+                List.of(primitiveJsonAsMap, primitiveJsonAsMap),
                 Utils.TABLE_COLUMN_METADATA,
                 fullMetadataJsonAsMap)),
         Arguments.of(
             "Empty nested array",
             buildRowWithDefaultMetadata("{\"key\": []}"),
             ImmutableMap.of(
-                "\"KEY\"", ImmutableList.of(), Utils.TABLE_COLUMN_METADATA, fullMetadataJsonAsMap)),
+                "\"KEY\"", List.of(), Utils.TABLE_COLUMN_METADATA, fullMetadataJsonAsMap)),
         Arguments.of(
             "Empty object",
             buildRowWithDefaultMetadata("{\"key\": {}}"),
@@ -339,15 +339,15 @@ private static Stream<Arguments> prepareNoSchematizationData() throws JsonProces
                 "{\"key\": [" + primitiveJsonExample + ", " + primitiveJsonExample + "]}"),
             ImmutableMap.of(
                 Utils.TABLE_COLUMN_CONTENT,
-                ImmutableMap.of("key", ImmutableList.of(primitiveJsonAsMap, primitiveJsonAsMap)),
+                ImmutableMap.of("key", List.of(primitiveJsonAsMap, primitiveJsonAsMap)),
                 Utils.TABLE_COLUMN_METADATA,
                 fullMetadataJsonAsMap)),
         Arguments.of(
             "Empty nested array",
             buildRowWithDefaultMetadata("{\"key\": []}"),
             ImmutableMap.of(
                 Utils.TABLE_COLUMN_CONTENT,
-                ImmutableMap.of("key", ImmutableList.of()),
+                ImmutableMap.of("key", List.of()),
                 Utils.TABLE_COLUMN_METADATA,
                 fullMetadataJsonAsMap)),
         Arguments.of(
diff --git a/src/test/java/com/snowflake/kafka/connector/streaming/iceberg/sql/ComplexJsonRecord.java b/src/test/java/com/snowflake/kafka/connector/streaming/iceberg/sql/ComplexJsonRecord.java
@@ -5,7 +5,6 @@
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.databind.ObjectMapper;
-import com.google.common.collect.ImmutableList;
 import com.snowflake.kafka.connector.Utils;
 import com.snowflake.kafka.connector.records.MetadataRecord;
 import java.io.IOException;
@@ -28,11 +27,11 @@ public class ComplexJsonRecord {
           0.5,
           0.25,
           true,
-          ImmutableList.of(1, 2, 3),
-          ImmutableList.of("a", "b", "c"),
-          ImmutableList.of(true),
-          ImmutableList.of(1, 4),
-          ImmutableList.of(ImmutableList.of(7, 8, 9), ImmutableList.of(10, 11, 12)),
+          List.of(1, 2, 3),
+          List.of("a", "b", "c"),
+          List.of(true),
+          List.of(1, 4),
+          List.of(List.of(7, 8, 9), List.of(10, 11, 12)),
           PrimitiveJsonRecord.primitiveJsonRecordValueExample,
           PrimitiveJsonRecord.primitiveJsonRecordValueExample);
 
diff --git a/src/test/java/net/snowflake/ingest/streaming/FakeSnowflakeStreamingIngestChannel.java b/src/test/java/net/snowflake/ingest/streaming/FakeSnowflakeStreamingIngestChannel.java
@@ -1,6 +1,5 @@
 package net.snowflake.ingest.streaming;
 
-import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Lists;
 import java.util.LinkedHashMap;
 import java.util.LinkedList;
@@ -160,6 +159,6 @@ public Map<String, ColumnProperties> getTableSchema() {
   }
 
   List<Map<String, Object>> getRows() {
-    return ImmutableList.copyOf(this.rows);
+    return List.copyOf(this.rows);
   }
 }

Original file line number	Diff line number	Diff line change
`@@ -1,6 +1,5 @@`
`1`	`1`	`package com.snowflake.kafka.connector.internal.streaming.schemaevolution;`
`2`	`2`
`3`		`-import com.google.common.collect.ImmutableList;`
`4`	`3`	`import com.google.common.collect.Iterables;`
`5`	`4`	`import com.google.common.collect.Lists;`
`6`	`5`	`import java.util.List;`
`@@ -21,10 +20,10 @@ public SchemaEvolutionTargetItems mapToSchemaEvolutionItems(`
`21`	`20`	`extraColNames);`
`22`	`21`	`}`
`23`	`22`
`24`		`- private List<String> joinNullableLists(List<String> list1, List<String> list2) {`
	`23`	`+ private static List<String> joinNullableLists(List<String> list1, List<String> list2) {`
`25`	`24`	`return Lists.newArrayList(`
`26`	`25`	`Iterables.concat(`
`27`		`- Optional.ofNullable(list1).orElse(ImmutableList.of()),`
`28`		`- Optional.ofNullable(list2).orElse(ImmutableList.of())));`
	`26`	`+ Optional.ofNullable(list1).orElse(List.of()),`
	`27`	`+ Optional.ofNullable(list2).orElse(List.of())));`
`29`	`28`	`}`
`30`	`29`	`}`