Merge remote-tracking branch 'origin/ENG-970-Schema_creation_issue' i…

…nto snowflake-v2.4.1-upgrade
streamkap-com · Nov 19, 2024 · a29ccc3 · a29ccc3
2 parents fc3154d + 493c23e
commit a29ccc3
Show file tree

Hide file tree

Showing 4 changed files with 20 additions and 14 deletions.
diff --git a/src/main/java/com/snowflake/kafka/connector/SnowflakeSinkConnector.java b/src/main/java/com/snowflake/kafka/connector/SnowflakeSinkConnector.java
@@ -294,18 +294,20 @@ public Config validate(Map<String, String> connectorConfigs) {
       return result;
     }
 
-    // Disabling config validation due to ENG-789/Auto create schema feature for Append SF connector
-    /*try {
-      testConnection.schemaExists(connectorConfigs.get(Utils.SF_SCHEMA));
-    } catch (SnowflakeKafkaConnectorException e) {
-      LOGGER.error("Validate Error msg:{}, errorCode:{}", e.getMessage(), e.getCode());
-      if (e.getCode().equals("2001")) {
-        Utils.updateConfigErrorMessage(result, Utils.SF_SCHEMA, " schema does not exist");
-      } else {
-        throw e;
+    boolean createSchemaAuto = Boolean.parseBoolean(connectorConfigs.getOrDefault(Utils.CREATE_SCHEMA_AUTO,"false"));
+    if(!createSchemaAuto) {
+      try {
+        testConnection.schemaExists(connectorConfigs.get(Utils.SF_SCHEMA));
+      } catch (SnowflakeKafkaConnectorException e) {
+        LOGGER.error("Validate Error msg:{}, errorCode:{}", e.getMessage(), e.getCode());
+        if (e.getCode().equals("2001")) {
+          Utils.updateConfigErrorMessage(result, Utils.SF_SCHEMA, " schema does not exist");
+        } else {
+          throw e;
+        }
+        return result;
       }
-      return result;
-    }*/
+    }
 
     LOGGER.info("Validated config with no error");
     return result;

diff --git a/src/main/java/com/snowflake/kafka/connector/SnowflakeSinkTask.java b/src/main/java/com/snowflake/kafka/connector/SnowflakeSinkTask.java
@@ -234,8 +234,11 @@ public void start(final Map<String, String> parsedConfig) {
             .setErrorReporter(kafkaRecordErrorReporter)
             .setSinkTaskContext(this.context)
             .build();
-    createSchemaIfNotExists(getConnection(),
-            parsedConfig.get(SnowflakeSinkConnectorConfig.SNOWFLAKE_SCHEMA));
+
+    if(Boolean.parseBoolean(parsedConfig.getOrDefault(Utils.CREATE_SCHEMA_AUTO,"false"))) {
+      createSchemaIfNotExists(getConnection(),
+              parsedConfig.get(SnowflakeSinkConnectorConfig.SNOWFLAKE_SCHEMA));
+    }
     this.streamkapQueryTemplate = StreamkapQueryTemplate.buildStreamkapQueryTemplateFromConfig(parsedConfig);
 
     DYNAMIC_LOGGER.info(

diff --git a/src/main/java/com/snowflake/kafka/connector/Utils.java b/src/main/java/com/snowflake/kafka/connector/Utils.java
@@ -136,6 +136,7 @@ public class Utils {
   public static final String TOPICS_MAP_CONF = "topics.config.map";
   public static final String SCHEMA_CHANGE_CHECK_MS = "schema.changes.check.interval.ms";
   public static final String APPLY_DYNAMIC_TABLE_SCRIPT_CONF = "apply.dynamic.table.script";
+  public static final String CREATE_SCHEMA_AUTO = "create.schema.auto";
 
   private static final KCLogger LOGGER = new KCLogger(Utils.class.getName());
 

diff --git a/src/main/java/com/snowflake/kafka/connector/internal/SnowflakeConnectionServiceV1.java b/src/main/java/com/snowflake/kafka/connector/internal/SnowflakeConnectionServiceV1.java
@@ -244,7 +244,7 @@ public void createStage(final String stageName) {
   public boolean schemaExist(final String schemaName) {
     checkConnection();
     InternalUtils.assertNotEmpty("schemaName", schemaName);
-    String query = "desc schema identifier(?)";
+    String query = "use schema identifier(?)";
     PreparedStatement stmt = null;
     boolean exist;
     try {