From ea009b55a385564f76beb99468f363edd3c4828a Mon Sep 17 00:00:00 2001
From: Y Ethan Guo <ethan.guoyihua@gmail.com>
Date: Tue, 30 Nov 2021 06:45:50 -0800
Subject: [PATCH] [HUDI-2891] Fix write configs for Java engine in Kafka
 Connect Sink (#4161)

---
 hudi-kafka-connect/demo/config-sink-hive.json                   | 1 -
 hudi-kafka-connect/demo/config-sink.json                        | 1 -
 .../apache/hudi/connect/writers/KafkaConnectWriterProvider.java | 2 ++
 .../java/org/apache/hudi/writers/TestBufferedConnectWriter.java | 2 ++
 4 files changed, 4 insertions(+), 2 deletions(-)

diff --git a/hudi-kafka-connect/demo/config-sink-hive.json b/hudi-kafka-connect/demo/config-sink-hive.json
index bf7e99833..214fd1891 100644
--- a/hudi-kafka-connect/demo/config-sink-hive.json
+++ b/hudi-kafka-connect/demo/config-sink-hive.json
@@ -10,7 +10,6 @@
 		"topics": "hudi-test-topic",
 		"hoodie.table.name": "hudi-test-topic",
 		"hoodie.table.type": "MERGE_ON_READ",
-		"hoodie.metadata.enable": "false",
 		"hoodie.base.path": "hdfs://namenode:8020/user/hive/warehouse/hudi-test-topic",
 		"hoodie.datasource.write.recordkey.field": "volume",
 		"hoodie.datasource.write.partitionpath.field": "date",
diff --git a/hudi-kafka-connect/demo/config-sink.json b/hudi-kafka-connect/demo/config-sink.json
index 9d1aedff6..105c4d491 100644
--- a/hudi-kafka-connect/demo/config-sink.json
+++ b/hudi-kafka-connect/demo/config-sink.json
@@ -10,7 +10,6 @@
 		"topics": "hudi-test-topic",
 		"hoodie.table.name": "hudi-test-topic",
 		"hoodie.table.type": "MERGE_ON_READ",
-		"hoodie.metadata.enable": "false",
 		"hoodie.base.path": "file:///tmp/hoodie/hudi-test-topic",
 		"hoodie.datasource.write.recordkey.field": "volume",
 		"hoodie.datasource.write.partitionpath.field": "date",
diff --git a/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/writers/KafkaConnectWriterProvider.java b/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/writers/KafkaConnectWriterProvider.java
index 18c1835f6..5bac1f077 100644
--- a/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/writers/KafkaConnectWriterProvider.java
+++ b/hudi-kafka-connect/src/main/java/org/apache/hudi/connect/writers/KafkaConnectWriterProvider.java
@@ -22,6 +22,7 @@ import org.apache.hudi.client.HoodieJavaWriteClient;
 import org.apache.hudi.client.WriteStatus;
 import org.apache.hudi.client.common.HoodieJavaEngineContext;
 import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.engine.EngineType;
 import org.apache.hudi.common.engine.HoodieEngineContext;
 import org.apache.hudi.common.model.HoodieAvroPayload;
 import org.apache.hudi.common.util.ReflectionUtils;
@@ -74,6 +75,7 @@ public class KafkaConnectWriterProvider implements ConnectWriterProvider<WriteSt
 
       // Create the write client to write some records in
       writeConfig = HoodieWriteConfig.newBuilder()
+          .withEngineType(EngineType.JAVA)
           .withProperties(connectConfigs.getProps())
           .withFileIdPrefixProviderClassName(KafkaConnectFileIdPrefixProvider.class.getName())
           .withProps(Collections.singletonMap(
diff --git a/hudi-kafka-connect/src/test/java/org/apache/hudi/writers/TestBufferedConnectWriter.java b/hudi-kafka-connect/src/test/java/org/apache/hudi/writers/TestBufferedConnectWriter.java
index b0dcf38f4..458c79a31 100644
--- a/hudi-kafka-connect/src/test/java/org/apache/hudi/writers/TestBufferedConnectWriter.java
+++ b/hudi-kafka-connect/src/test/java/org/apache/hudi/writers/TestBufferedConnectWriter.java
@@ -20,6 +20,7 @@ package org.apache.hudi.writers;
 
 import org.apache.hudi.client.HoodieJavaWriteClient;
 import org.apache.hudi.client.common.HoodieJavaEngineContext;
+import org.apache.hudi.common.engine.EngineType;
 import org.apache.hudi.common.model.HoodieRecord;
 import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
 import org.apache.hudi.common.util.Option;
@@ -62,6 +63,7 @@ public class TestBufferedConnectWriter {
     configs = KafkaConnectConfigs.newBuilder().build();
     schemaProvider = new TestAbstractConnectWriter.TestSchemaProvider();
     writeConfig = HoodieWriteConfig.newBuilder()
+        .withEngineType(EngineType.JAVA)
         .withPath("/tmp")
         .withSchema(schemaProvider.getSourceSchema().toString())
         .build();