From 7b1452b58ef2e8ef4774afb662c25c07da696dd5 Mon Sep 17 00:00:00 2001
From: Wojciech Trefon <wojciech.trefon@snowflake.com>
Date: Mon, 7 Oct 2024 16:12:14 +0200
Subject: [PATCH] CR changes

---
 .../config/IcebergConfigValidationTest.java   |  4 +--
 .../IcebergTableSchemaValidatorIT.java        | 31 ++++++-------------
 2 files changed, 10 insertions(+), 25 deletions(-)

diff --git a/src/test/java/com/snowflake/kafka/connector/config/IcebergConfigValidationTest.java b/src/test/java/com/snowflake/kafka/connector/config/IcebergConfigValidationTest.java
index 147dfee3a..6bcd570d0 100644
--- a/src/test/java/com/snowflake/kafka/connector/config/IcebergConfigValidationTest.java
+++ b/src/test/java/com/snowflake/kafka/connector/config/IcebergConfigValidationTest.java
@@ -1,6 +1,5 @@
 package com.snowflake.kafka.connector.config;
 
-import static com.snowflake.kafka.connector.SnowflakeSinkConnectorConfig.ENABLE_SCHEMATIZATION_CONFIG;
 import static com.snowflake.kafka.connector.SnowflakeSinkConnectorConfig.INGESTION_METHOD_OPT;
 
 import com.google.common.collect.ImmutableMap;
@@ -44,8 +43,7 @@ public static Stream<Arguments> validConfigs() {
         Arguments.of(
             SnowflakeSinkConnectorConfigBuilder.icebergConfig()
                 .withSchematizationEnabled(false)
-                .build(),
-            ENABLE_SCHEMATIZATION_CONFIG));
+                .build()));
   }
 
   public static Stream<Arguments> invalidConfigs() {
diff --git a/src/test/java/com/snowflake/kafka/connector/streaming/iceberg/IcebergTableSchemaValidatorIT.java b/src/test/java/com/snowflake/kafka/connector/streaming/iceberg/IcebergTableSchemaValidatorIT.java
index 05af8350e..2f659e5e0 100644
--- a/src/test/java/com/snowflake/kafka/connector/streaming/iceberg/IcebergTableSchemaValidatorIT.java
+++ b/src/test/java/com/snowflake/kafka/connector/streaming/iceberg/IcebergTableSchemaValidatorIT.java
@@ -8,6 +8,8 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 
 public class IcebergTableSchemaValidatorIT extends BaseIcebergIT {
 
@@ -48,13 +50,6 @@ public void shouldValidateExpectedIcebergTableSchema() {
       schemaValidator.validateTable(tableName, TEST_ROLE, SCHEMA_EVOLUTION);
     }
 
-    @Test
-    public void shouldThrowExceptionWhenTableDoesNotExist() {
-      Assertions.assertThrows(
-          SnowflakeKafkaConnectorException.class,
-          () -> schemaValidator.validateTable(tableName, TEST_ROLE, SCHEMA_EVOLUTION));
-    }
-
     @Test
     public void shouldNotThrowExceptionWhenColumnRecordMetadataDoesNotExist() {
       // given
@@ -86,7 +81,6 @@ class SchemaEvolutionNotEnabled {
     public void shouldValidateExpectedIcebergTableSchema() {
       // given
       createIcebergTableNoSchemaEvolution(tableName);
-      enableSchemaEvolution(tableName);
 
       // when, then
       schemaValidator.validateTable(tableName, TEST_ROLE, SCHEMA_EVOLUTION);
@@ -103,7 +97,6 @@ public void shouldThrowExceptionWhenTableDoesNotExist() {
     public void shouldThrowExceptionWhenRecordContentDoesNotExist() {
       // given
       createIcebergTableWithColumnClause(tableName, "some_column VARCHAR");
-      enableSchemaEvolution(tableName);
 
       // expect
       Assertions.assertThrows(
@@ -115,7 +108,6 @@ public void shouldThrowExceptionWhenRecordContentDoesNotExist() {
     public void shouldThrowExceptionWhenRecordContentHasInvalidType() {
       // given
       createIcebergTableWithColumnClause(tableName, "record_content MAP(VARCHAR, VARCHAR)");
-      enableSchemaEvolution(tableName);
 
       // expect
       Assertions.assertThrows(
@@ -127,22 +119,17 @@ public void shouldThrowExceptionWhenRecordContentHasInvalidType() {
     public void shouldNotThrowExceptionWhenColumnRecordMetadataDoesNotExist() {
       // given
       createIcebergTableWithColumnClause(tableName, "record_content object()");
-      enableSchemaEvolution(tableName);
 
       // expect
       schemaValidator.validateTable(tableName, TEST_ROLE, SCHEMA_EVOLUTION);
     }
+  }
 
-    @Test
-    public void shouldThrowExceptionWhenRecordMetadataHasInvalidType() {
-      // given
-      createIcebergTableWithColumnClause(tableName, "record_metadata MAP(VARCHAR, VARCHAR)");
-      enableSchemaEvolution(tableName);
-
-      // expect
-      Assertions.assertThrows(
-          SnowflakeKafkaConnectorException.class,
-          () -> schemaValidator.validateTable(tableName, TEST_ROLE, SCHEMA_EVOLUTION));
-    }
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
+  public void shouldThrowExceptionWhenTableDoesNotExist(boolean schemaEvolution) {
+    Assertions.assertThrows(
+        SnowflakeKafkaConnectorException.class,
+        () -> schemaValidator.validateTable(tableName, TEST_ROLE, schemaEvolution));
   }
 }