FLOW-7155 Adapt iceberg tests for ssv2

Author: sfc-gh-lkucharski

pom.xml

@@ -594,7 +594,7 @@
         <dependency>
             <groupId>org.testcontainers</groupId>
             <artifactId>testcontainers</artifactId>
-            <version>1.19.3</version>
+            <version>2.0.2</version>
             <scope>test</scope>
         </dependency>
 

src/main/java/com/snowflake/kafka/connector/SemanticVersion.java

@@ -5,14 +5,14 @@
 import java.util.regex.Pattern;
 
 /** Represents a parsed semantic version. */
-class SemanticVersion implements Comparable<SemanticVersion> {
-  final int major;
-  final int minor;
-  final int patch;
-  final boolean isReleaseCandidate;
-  final String originalVersion;
-
-  SemanticVersion(String version) {
+public class SemanticVersion implements Comparable<SemanticVersion> {
+  private final int major;
+  private final int minor;
+  private final int patch;
+  private final boolean isReleaseCandidate;
+  private final String originalVersion;
+
+  public SemanticVersion(String version) {
     this.originalVersion = version;
     // Pattern to match versions like "3.1.0" or "4.0.0-rc" or "4.0.0-RC1"
     Pattern pattern = Pattern.compile("(\\d+)\\.(\\d+)\\.(\\d+)(?:-[rR][cC]\\d*)?");
@@ -26,6 +26,26 @@ class SemanticVersion implements Comparable<SemanticVersion> {
     this.isReleaseCandidate = version.toLowerCase().contains("-rc");
   }
 
+  public String originalVersion() {
+    return originalVersion;
+  }
+
+  public boolean isReleaseCandidate() {
+    return isReleaseCandidate;
+  }
+
+  public int major() {
+    return major;
+  }
+
+  public int minor() {
+    return minor;
+  }
+
+  public int patch() {
+    return patch;
+  }
+
   @Override
   public int compareTo(SemanticVersion other) {
     if (this.major != other.major) {

src/main/java/com/snowflake/kafka/connector/SnowflakeSinkTask.java

@@ -346,7 +346,7 @@ public String version() {
    * @param config connector config file
    * @return result map
    */
-  static Map<String, String> getTopicToTableMap(Map<String, String> config) {
+  public static Map<String, String> getTopicToTableMap(Map<String, String> config) {
     if (config.containsKey(KafkaConnectorConfigParams.SNOWFLAKE_TOPICS2TABLE_MAP)) {
       Map<String, String> result =
           Utils.parseTopicToTableMap(

src/main/java/com/snowflake/kafka/connector/Utils.java

@@ -94,7 +94,7 @@ public static String quoteNameIfNeeded(String name) {
    *   <li>Never suggest RC (release candidate) versions
    * </ul>
    */
-  static boolean checkConnectorVersion() {
+  public static boolean checkConnectorVersion() {
     return checkConnectorVersion(VERSION, fetchAvailableVersionsFromMaven());
   }
 
@@ -172,7 +172,7 @@ static String findRecommendedVersion(
         SemanticVersion version = new SemanticVersion(versionString);
 
         // Skip RC versions
-        if (version.isReleaseCandidate) {
+        if (version.isReleaseCandidate()) {
           continue;
         }
 
@@ -198,7 +198,7 @@ static String findRecommendedVersion(
    *
    * @param config connector configuration
    */
-  static ImmutableMap<String, String> validateProxySettings(Map<String, String> config) {
+  public static ImmutableMap<String, String> validateProxySettings(Map<String, String> config) {
     Map<String, String> invalidConfigParams = new HashMap<String, String>();
 
     String host =
@@ -238,7 +238,7 @@ static ImmutableMap<String, String> validateProxySettings(Map<String, String> co
    *
    * @param config connector configuration
    */
-  static void enableJVMProxy(Map<String, String> config) {
+  public static void enableJVMProxy(Map<String, String> config) {
     String host =
         ConnectorConfigTools.getProperty(config, KafkaConnectorConfigParams.JVM_PROXY_HOST);
     String port =
@@ -307,7 +307,7 @@ static boolean isValidSnowflakeObjectIdentifier(String objName) {
    * @param appName snowflake application name
    * @return true if given application name is valid
    */
-  static boolean isValidSnowflakeApplicationName(String appName) {
+  public static boolean isValidSnowflakeApplicationName(String appName) {
     return appName.matches("^[-_a-zA-Z]{1}[-_$a-zA-Z0-9]+$");
   }
 

src/main/java/com/snowflake/kafka/connector/internal/DescribeTableRow.java

@@ -30,6 +30,6 @@ public int hashCode() {
 
   @Override
   public String toString() {
-    return "DescribeTableRow{" + "column='" + column + '\'' + ", type='" + type + '\'' + '}';
+    return " " + column + " " + type;
   }
 }

src/test/java/com/snowflake/kafka/connector/ConnectorConfigValidatorTest.java

@@ -17,7 +17,6 @@
 import static com.snowflake.kafka.connector.Constants.KafkaConnectorConfigParams.SNOWFLAKE_PRIVATE_KEY;
 import static com.snowflake.kafka.connector.Constants.KafkaConnectorConfigParams.SNOWFLAKE_ROLE_NAME;
 import static com.snowflake.kafka.connector.Constants.KafkaConnectorConfigParams.SNOWFLAKE_SCHEMA_NAME;
-import static com.snowflake.kafka.connector.Constants.KafkaConnectorConfigParams.SNOWFLAKE_STREAMING_ICEBERG_ENABLED;
 import static com.snowflake.kafka.connector.Constants.KafkaConnectorConfigParams.SNOWFLAKE_STREAMING_MAX_CLIENT_LAG;
 import static com.snowflake.kafka.connector.Constants.KafkaConnectorConfigParams.SNOWFLAKE_TOPICS2TABLE_MAP;
 import static com.snowflake.kafka.connector.Constants.KafkaConnectorConfigParams.SNOWFLAKE_URL_NAME;
@@ -62,31 +61,13 @@ public static Stream<Arguments> validConfigs() {
         Arguments.of(SnowflakeSinkConnectorConfigBuilder.streamingConfig().build()));
   }
 
-  static Stream<Arguments> invalidConfigs() {
-    return Stream.of(
-        Arguments.of(
-            SnowflakeSinkConnectorConfigBuilder.icebergConfig().withIcebergEnabled().build(),
-            "snowflake.streaming.iceberg.enabled"),
-        Arguments.of(
-            SnowflakeSinkConnectorConfigBuilder.icebergConfig().build(),
-            "snowflake.streaming.iceberg.enabled"));
-  }
-
   @ParameterizedTest(name = "Valid config: {0}")
   @MethodSource("validConfigs")
   public void shouldValidateCorrectConfig(Map<String, String> config) {
     // no exception thrown
     connectorConfigValidator.validateConfig(config);
   }
 
-  @ParameterizedTest(name = "Invalid config: {0}")
-  @MethodSource("invalidConfigs")
-  void shouldReturnErrorOnInvalidConfig(final Map<String, String> config, String errorKey) {
-    assertThatThrownBy(() -> connectorConfigValidator.validateConfig(config))
-        .isInstanceOf(SnowflakeKafkaConnectorException.class)
-        .hasMessageContaining(errorKey);
-  }
-
   @Test
   public void testConfig() {
     Map<String, String> config = SnowflakeSinkConnectorConfigBuilder.streamingConfig().build();
@@ -480,25 +461,6 @@ public void shouldThrowExceptionWhenRoleNotDefinedForSSv2() {
         .hasMessageContaining(SNOWFLAKE_ROLE_NAME);
   }
 
-  @Test
-  public void shouldThrowExceptionWhenBothSSv2AndIcebergEnabled() {
-    Map<String, String> config =
-        SnowflakeSinkConnectorConfigBuilder.streamingConfig().withIcebergEnabled().build();
-
-    assertThatThrownBy(() -> connectorConfigValidator.validateConfig(config))
-        .isInstanceOf(SnowflakeKafkaConnectorException.class)
-        .hasMessageContaining("Ingestion to Iceberg table is currently unsupported")
-        .hasMessageContaining(SNOWFLAKE_STREAMING_ICEBERG_ENABLED);
-  }
-
-  @Test
-  public void shouldValidateSSv2WithoutIceberg() {
-    Map<String, String> config = SnowflakeSinkConnectorConfigBuilder.streamingConfig().build();
-
-    assertThatCode(() -> connectorConfigValidator.validateConfig(config))
-        .doesNotThrowAnyException();
-  }
-
   private void invalidConfigRunner(List<String> paramsToRemove) {
     Map<String, String> config = getConfig();
     for (String configParam : paramsToRemove) {