[HUDI-2394] Implement Kafka Sink Protocol for Hudi for Ingesting Immutable Data (apache#3592)

 - Fixing packaging, naming of classes
 - Use of log4j over slf4j for uniformity
- More follow-on fixes
 - Added a version to control/coordinator events.
 - Eliminated the config added to write config
 - Fixed fetching of checkpoints based on table type
 - Clean up of naming, code placement

Co-authored-by: Rajesh Mahindra <[email protected]>
Co-authored-by: Vinoth Chandar <[email protected]>

Author: 3 people

.gitignore

@@ -78,4 +78,4 @@ dependency-reduced-pom.xml
 #######################################
 hudi-integ-test/compose_env
 node_modules
-package-lock.json
+package-lock.json

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java

@@ -46,6 +46,7 @@
 import org.apache.hudi.keygen.constant.KeyGeneratorType;
 import org.apache.hudi.metrics.MetricsReporterType;
 import org.apache.hudi.metrics.datadog.DatadogHttpClient.ApiSite;
+import org.apache.hudi.table.RandomFileIdPrefixProvider;
 import org.apache.hudi.table.action.compact.CompactionTriggerStrategy;
 import org.apache.hudi.table.action.compact.strategy.CompactionStrategy;
 
@@ -413,6 +414,12 @@ public class HoodieWriteConfig extends HoodieConfig {
       .withDocumentation("Whether to include '_hoodie_operation' in the metadata fields. "
           + "Once enabled, all the changes of a record are persisted to the delta log directly without merge");
 
+  public static final ConfigProperty<String> FILEID_PREFIX_PROVIDER_CLASS = ConfigProperty
+      .key("hoodie.fileid.prefix.provider.class")
+      .defaultValue(RandomFileIdPrefixProvider.class.getName())
+      .sinceVersion("0.10.0")
+      .withDocumentation("File Id Prefix provider class, that implements `org.apache.hudi.fileid.FileIdPrefixProvider`");
+
   private ConsistencyGuardConfig consistencyGuardConfig;
 
   // Hoodie Write Client transparently rewrites File System View config when embedded mode is enabled
@@ -1748,6 +1755,10 @@ public boolean allowOperationMetadataField() {
     return getBooleanOrDefault(ALLOW_OPERATION_METADATA_FIELD);
   }
 
+  public String getFileIdPrefixProviderClassName() {
+    return getString(FILEID_PREFIX_PROVIDER_CLASS);
+  }
+
   public static class Builder {
 
     protected final HoodieWriteConfig writeConfig = new HoodieWriteConfig();
@@ -2079,6 +2090,11 @@ public Builder withAllowOperationMetadataField(boolean allowOperationMetadataFie
       return this;
     }
 
+    public Builder withFileIdPrefixProviderClassName(String fileIdPrefixProviderClassName) {
+      writeConfig.setValue(FILEID_PREFIX_PROVIDER_CLASS, fileIdPrefixProviderClassName);
+      return this;
+    }
+
     public Builder withProperties(Properties properties) {
       this.writeConfig.getProps().putAll(properties);
       return this;

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/FileIdPrefixProvider.java

@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.table;
+
+import java.util.Properties;
+
+public abstract class FileIdPrefixProvider {
+
+  private final Properties props;
+
+  public FileIdPrefixProvider(Properties props) {
+    this.props = props;
+  }
+
+  public Properties getProps() {
+    return props;
+  }
+
+  public abstract String createFilePrefix(String partitionPath);
+}

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/RandomFileIdPrefixProvider.java

@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.table;
+
+import org.apache.hudi.common.fs.FSUtils;
+
+import java.util.Properties;
+
+public class RandomFileIdPrefixProvider extends FileIdPrefixProvider {
+
+  public RandomFileIdPrefixProvider(Properties props) {
+    super(props);
+  }
+
+  @Override
+  public String createFilePrefix(String partitionPath) {
+    return FSUtils.createNewFileIdPfx();
+  }
+}

hudi-client/hudi-java-client/src/main/java/org/apache/hudi/client/HoodieJavaWriteClient.java

@@ -18,8 +18,6 @@
 
 package org.apache.hudi.client;
 
-import com.codahale.metrics.Timer;
-import org.apache.hadoop.conf.Configuration;
 import org.apache.hudi.client.common.HoodieJavaEngineContext;
 import org.apache.hudi.client.embedded.EmbeddedTimelineService;
 import org.apache.hudi.common.engine.HoodieEngineContext;
@@ -30,6 +28,7 @@
 import org.apache.hudi.common.model.HoodieWriteStat;
 import org.apache.hudi.common.model.WriteOperationType;
 import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
 import org.apache.hudi.common.table.timeline.HoodieTimeline;
 import org.apache.hudi.common.util.Option;
 import org.apache.hudi.config.HoodieWriteConfig;
@@ -41,6 +40,9 @@
 import org.apache.hudi.table.HoodieTable;
 import org.apache.hudi.table.action.HoodieWriteMetadata;
 
+import com.codahale.metrics.Timer;
+import org.apache.hadoop.conf.Configuration;
+
 import java.io.IOException;
 import java.util.List;
 import java.util.Map;
@@ -153,11 +155,23 @@ public List<WriteStatus> bulkInsert(List<HoodieRecord<T>> records,
     throw new HoodieNotSupportedException("BulkInsert is not supported in HoodieJavaClient");
   }
 
+  public void transitionInflight(String instantTime) {
+    HoodieTableMetaClient metaClient = createMetaClient(true);
+    metaClient.getActiveTimeline().transitionRequestedToInflight(
+        new HoodieInstant(HoodieInstant.State.REQUESTED, metaClient.getCommitActionType(), instantTime),
+        Option.empty(), config.shouldAllowMultiWriteOnSameInstant());
+  }
+
   @Override
   public List<WriteStatus> bulkInsertPreppedRecords(List<HoodieRecord<T>> preppedRecords,
                                                     String instantTime,
                                                     Option<BulkInsertPartitioner<List<HoodieRecord<T>>>> bulkInsertPartitioner) {
-    throw new HoodieNotSupportedException("BulkInsertPreppedRecords is not supported in HoodieJavaClient");
+    HoodieTable<T, List<HoodieRecord<T>>, List<HoodieKey>, List<WriteStatus>> table =
+        getTableAndInitCtx(WriteOperationType.BULK_INSERT_PREPPED, instantTime);
+    table.validateInsertSchema();
+    preWrite(instantTime, WriteOperationType.BULK_INSERT_PREPPED, table.getMetaClient());
+    HoodieWriteMetadata<List<WriteStatus>> result = table.bulkInsertPrepped(context, instantTime, preppedRecords, bulkInsertPartitioner);
+    return postWrite(result, instantTime, table);
   }
 
   @Override

hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/action/commit/JavaBulkInsertHelper.java

@@ -19,17 +19,18 @@
 package org.apache.hudi.table.action.commit;
 
 import org.apache.hudi.client.WriteStatus;
-import org.apache.hudi.common.fs.FSUtils;
 import org.apache.hudi.common.model.HoodieKey;
 import org.apache.hudi.common.model.HoodieRecord;
 import org.apache.hudi.common.model.HoodieRecordPayload;
 import org.apache.hudi.common.table.timeline.HoodieInstant;
 import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.ReflectionUtils;
 import org.apache.hudi.config.HoodieWriteConfig;
 import org.apache.hudi.execution.JavaLazyInsertIterable;
 import org.apache.hudi.execution.bulkinsert.JavaBulkInsertInternalPartitionerFactory;
 import org.apache.hudi.io.CreateHandleFactory;
 import org.apache.hudi.table.BulkInsertPartitioner;
+import org.apache.hudi.table.FileIdPrefixProvider;
 import org.apache.hudi.table.HoodieTable;
 import org.apache.hudi.table.action.HoodieWriteMetadata;
 
@@ -66,10 +67,14 @@ public HoodieWriteMetadata<List<WriteStatus>> bulkInsert(final List<HoodieRecord
                                                            final Option<BulkInsertPartitioner<T>> userDefinedBulkInsertPartitioner) {
     HoodieWriteMetadata result = new HoodieWriteMetadata();
 
-    //transition bulk_insert state to inflight
-    table.getActiveTimeline().transitionRequestedToInflight(new HoodieInstant(HoodieInstant.State.REQUESTED,
-            table.getMetaClient().getCommitActionType(), instantTime), Option.empty(),
-        config.shouldAllowMultiWriteOnSameInstant());
+    // It's possible the transition to inflight could have already happened.
+    if (!table.getActiveTimeline().filterInflights().containsInstant(instantTime)) {
+      table.getActiveTimeline().transitionRequestedToInflight(
+          new HoodieInstant(HoodieInstant.State.REQUESTED, table.getMetaClient().getCommitActionType(), instantTime),
+          Option.empty(),
+          config.shouldAllowMultiWriteOnSameInstant());
+    }
+
     // write new files
     List<WriteStatus> writeStatuses = bulkInsert(inputRecords, instantTime, table, config, performDedupe, userDefinedBulkInsertPartitioner, false, config.getBulkInsertShuffleParallelism(), false);
     //update index
@@ -102,12 +107,16 @@ public List<WriteStatus> bulkInsert(List<HoodieRecord<T>> inputRecords,
         : JavaBulkInsertInternalPartitionerFactory.get(config.getBulkInsertSortMode());
     repartitionedRecords = (List<HoodieRecord<T>>) partitioner.repartitionRecords(dedupedRecords, parallelism);
 
-    String idPfx = FSUtils.createNewFileIdPfx();
+    FileIdPrefixProvider fileIdPrefixProvider = (FileIdPrefixProvider) ReflectionUtils.loadClass(
+        config.getFileIdPrefixProviderClassName(),
+        config.getProps());
 
     List<WriteStatus> writeStatuses = new ArrayList<>();
 
-    new JavaLazyInsertIterable<>(repartitionedRecords.iterator(), true, config, instantTime, table, idPfx,
-        table.getTaskContextSupplier(), new CreateHandleFactory<>()).forEachRemaining(writeStatuses::addAll);
+    new JavaLazyInsertIterable<>(repartitionedRecords.iterator(), true,
+        config, instantTime, table,
+        fileIdPrefixProvider.createFilePrefix(""), table.getTaskContextSupplier(),
+        new CreateHandleFactory<>()).forEachRemaining(writeStatuses::addAll);
 
     return writeStatuses;
   }

hudi-common/src/main/java/org/apache/hudi/common/config/ConfigGroups.java

@@ -30,7 +30,8 @@ public enum Names {
     FLINK_SQL("Flink Sql Configs"),
     WRITE_CLIENT("Write Client Configs"),
     METRICS("Metrics Configs"),
-    RECORD_PAYLOAD("Record Payload Config");
+    RECORD_PAYLOAD("Record Payload Config"),
+    KAFKA_CONNECT("Kafka Connect Configs");
 
     public final String name;
 
@@ -72,6 +73,9 @@ public static String getDescription(Names names) {
         description = "These set of configs are used to enable monitoring and reporting of key"
             + "Hudi stats and metrics.";
         break;
+      case KAFKA_CONNECT:
+        description = "These set of configs are used for Kakfa Connect Sink Connector for writing Hudi Tables";
+        break;
       default:
         description = "Please fill in the description for Config Group Name: " + names.name;
         break;

hudi-common/src/main/java/org/apache/hudi/common/util/CommitUtils.java

@@ -26,6 +26,7 @@
 import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
 import org.apache.hudi.common.table.timeline.HoodieTimeline;
 import org.apache.hudi.exception.HoodieException;
+
 import org.apache.log4j.LogManager;
 import org.apache.log4j.Logger;
 

hudi-common/src/test/java/org/apache/hudi/common/testutils/SchemaTestUtil.java

@@ -71,17 +71,20 @@ public static List<IndexedRecord> generateTestRecords(int from, int limit) throw
     return toRecords(getSimpleSchema(), getSimpleSchema(), from, limit);
   }
 
+  public static List<String> generateTestJsonRecords(int from, int limit) throws IOException, URISyntaxException {
+    Path dataPath = initializeSampleDataPath();
+
+    try (Stream<String> stream = Files.lines(dataPath)) {
+      return stream.skip(from).limit(limit).collect(Collectors.toList());
+    } catch (IOException e) {
+      throw new HoodieIOException("Could not read data from " + RESOURCE_SAMPLE_DATA, e);
+    }
+  }
+
   private static List<IndexedRecord> toRecords(Schema writerSchema, Schema readerSchema, int from, int limit)
       throws IOException, URISyntaxException {
     GenericDatumReader<IndexedRecord> reader = new GenericDatumReader<>(writerSchema, readerSchema);
-    // Required to register the necessary JAR:// file system
-    URI resource = SchemaTestUtil.class.getResource(RESOURCE_SAMPLE_DATA).toURI();
-    Path dataPath;
-    if (resource.toString().contains("!")) {
-      dataPath = uriToPath(resource);
-    } else {
-      dataPath = Paths.get(SchemaTestUtil.class.getResource(RESOURCE_SAMPLE_DATA).toURI());
-    }
+    Path dataPath = initializeSampleDataPath();
 
     try (Stream<String> stream = Files.lines(dataPath)) {
       return stream.skip(from).limit(limit).map(s -> {
@@ -96,6 +99,21 @@ private static List<IndexedRecord> toRecords(Schema writerSchema, Schema readerS
     }
   }
 
+  /**
+   * Required to register the necessary JAR:// file system.
+   * @return Path to the sample data in the resource file.
+   * @throws IOException
+   * @throws URISyntaxException
+   */
+  private static Path initializeSampleDataPath() throws IOException, URISyntaxException {
+    URI resource = SchemaTestUtil.class.getResource(RESOURCE_SAMPLE_DATA).toURI();
+    if (resource.toString().contains("!")) {
+      return uriToPath(resource);
+    } else {
+      return Paths.get(SchemaTestUtil.class.getResource(RESOURCE_SAMPLE_DATA).toURI());
+    }
+  }
+
   public static Path uriToPath(URI uri) throws IOException {
     final Map<String, String> env = new HashMap<>();
     final String[] array = uri.toString().split("!");