refactor: Add Lombok annotations to hudi-utilities (Part 3)

Author: voonhous

hudi-utilities/pom.xml

@@ -230,6 +230,12 @@
       <artifactId>log4j-1.2-api</artifactId>
     </dependency>
 
+    <!-- Lombok -->
+    <dependency>
+      <groupId>org.projectlombok</groupId>
+      <artifactId>lombok</artifactId>
+    </dependency>
+
     <!-- Fasterxml -->
     <dependency>
       <groupId>com.fasterxml.jackson.module</groupId>

hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/BootstrapExecutor.java

@@ -44,12 +44,12 @@
 import org.apache.hudi.utilities.UtilHelpers;
 import org.apache.hudi.utilities.schema.SchemaProvider;
 
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.spark.api.java.JavaSparkContext;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.io.Serializable;
@@ -71,10 +71,9 @@
 /**
  * Performs bootstrap from a non-hudi source.
  */
+@Slf4j
 public class BootstrapExecutor implements Serializable {
 
-  private static final Logger LOG = LoggerFactory.getLogger(BootstrapExecutor.class);
-
   /**
    * Config.
    */
@@ -103,6 +102,7 @@ public class BootstrapExecutor implements Serializable {
   /**
    * Bootstrap Configuration.
    */
+  @Getter
   private final HoodieWriteConfig bootstrapConfig;
 
   /**
@@ -146,7 +146,7 @@ public BootstrapExecutor(HoodieStreamer.Config cfg, JavaSparkContext jssc, FileS
       builder = builder.withSchema(schemaProvider.getTargetHoodieSchema().toString());
     }
     this.bootstrapConfig = builder.build();
-    LOG.info("Created bootstrap executor with configs : " + bootstrapConfig.getProps());
+    log.info("Created bootstrap executor with configs: {}", bootstrapConfig.getProps());
   }
 
   /**
@@ -189,7 +189,7 @@ private void initializeTable() throws IOException {
     Path basePath = new Path(cfg.targetBasePath);
     if (fs.exists(basePath)) {
       if (cfg.bootstrapOverwrite) {
-        LOG.info("Target base path already exists, overwrite it");
+        log.info("Target base path already exists, overwrite it");
         fs.delete(basePath, true);
       } else {
         throw new HoodieException("target base path already exists at " + cfg.targetBasePath
@@ -244,8 +244,4 @@ private void initializeTable() throws IOException {
 
     builder.initTable(HadoopFSUtils.getStorageConfWithCopy(jssc.hadoopConfiguration()), cfg.targetBasePath);
   }
-
-  public HoodieWriteConfig getBootstrapConfig() {
-    return bootstrapConfig;
-  }
 }

hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/DefaultStreamContext.java

@@ -21,28 +21,18 @@
 import org.apache.hudi.common.util.Option;
 import org.apache.hudi.utilities.schema.SchemaProvider;
 
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+
 /**
  * The default implementation for the StreamContext interface,
  * composes SchemaProvider and SourceProfileSupplier currently,
  * can be extended for other arguments in the future.
  */
+@AllArgsConstructor
+@Getter
 public class DefaultStreamContext implements StreamContext {
 
   private final SchemaProvider schemaProvider;
   private final Option<SourceProfileSupplier> sourceProfileSupplier;
-
-  public DefaultStreamContext(SchemaProvider schemaProvider, Option<SourceProfileSupplier> sourceProfileSupplier) {
-    this.schemaProvider = schemaProvider;
-    this.sourceProfileSupplier = sourceProfileSupplier;
-  }
-
-  @Override
-  public SchemaProvider getSchemaProvider() {
-    return schemaProvider;
-  }
-
-  @Override
-  public Option<SourceProfileSupplier> getSourceProfileSupplier() {
-    return sourceProfileSupplier;
-  }
 }

hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/ErrorEvent.java

@@ -19,45 +19,16 @@
 
 package org.apache.hudi.utilities.streamer;
 
-import java.util.Objects;
+import lombok.Value;
 
 /**
  * Error event is an event triggered during write or processing failure of a record.
  */
+@Value
 public class ErrorEvent<T> {
 
-  private final ErrorReason reason;
-  private final T payload;
-
-  public ErrorEvent(T payload, ErrorReason reason) {
-    this.payload = payload;
-    this.reason = reason;
-  }
-
-  public T getPayload() {
-    return payload;
-  }
-
-  public ErrorReason getReason() {
-    return reason;
-  }
-
-  @Override
-  public boolean equals(Object o) {
-    if (this == o) {
-      return true;
-    }
-    if (o == null || getClass() != o.getClass()) {
-      return false;
-    }
-    ErrorEvent<?> that = (ErrorEvent<?>) o;
-    return reason == that.reason && Objects.equals(payload, that.payload);
-  }
-
-  @Override
-  public int hashCode() {
-    return Objects.hash(reason, payload);
-  }
+  T payload;
+  ErrorReason reason;
 
   /**
    * The reason behind write or processing failure of a record

hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieMultiTableStreamer.java

@@ -39,12 +39,13 @@
 
 import com.beust.jcommander.JCommander;
 import com.beust.jcommander.Parameter;
+import lombok.AccessLevel;
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.FileUtil;
 import org.apache.hadoop.fs.Path;
 import org.apache.spark.api.java.JavaSparkContext;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.io.Serializable;
@@ -69,11 +70,12 @@
  * Helps with ingesting incremental data into hoodie datasets for multiple tables.
  * Supports COPY_ON_WRITE and MERGE_ON_READ storage types.
  */
+@Getter
+@Slf4j
 public class HoodieMultiTableStreamer {
 
-  private static final Logger LOG = LoggerFactory.getLogger(HoodieMultiTableStreamer.class);
-
   private final List<TableExecutionContext> tableExecutionContexts;
+  @Getter(AccessLevel.NONE)
   private transient JavaSparkContext jssc;
   private final Set<String> successTables;
   private final Set<String> failedTables;
@@ -120,7 +122,7 @@ private void checkIfTableConfigFileExists(String configFolder, FileSystem fs, St
   //commonProps are passed as parameter which contain table to config file mapping
   private void populateTableExecutionContextList(TypedProperties properties, String configFolder, FileSystem fs, Config config) throws IOException {
     List<String> tablesToBeIngested = getTablesToBeIngested(properties);
-    LOG.info("tables to be ingested via MultiTableDeltaStreamer : " + tablesToBeIngested);
+    log.info("tables to be ingested via MultiTableDeltaStreamer : {}", tablesToBeIngested);
     TableExecutionContext executionContext;
     for (String table : tablesToBeIngested) {
       String[] tableWithDatabase = table.split("\\.");
@@ -271,11 +273,11 @@ public static void main(String[] args) throws IOException {
     }
 
     if (config.enableHiveSync) {
-      LOG.warn("--enable-hive-sync will be deprecated in a future release; please use --enable-sync instead for Hive syncing");
+      log.warn("--enable-hive-sync will be deprecated in a future release; please use --enable-sync instead for Hive syncing");
     }
 
     if (config.targetTableName != null) {
-      LOG.warn("--target-table is deprecated and will be removed in a future release due to it's useless;"
+      log.warn("--target-table is deprecated and will be removed in a future release due to it's useless;"
           + " please use {} to configure multiple target tables", HoodieStreamerConfig.TABLES_TO_BE_INGESTED.key());
     }
 
@@ -466,7 +468,7 @@ public void sync() {
         successTables.add(Helpers.getTableWithDatabase(context));
         streamer.shutdownGracefully();
       } catch (Exception e) {
-        LOG.error("error while running MultiTableDeltaStreamer for table: " + context.getTableName(), e);
+        log.error("error while running MultiTableDeltaStreamer for table: {}", context.getTableName(), e);
         failedTables.add(Helpers.getTableWithDatabase(context));
       } finally {
         if (streamer != null) {
@@ -475,9 +477,9 @@ public void sync() {
       }
     }
 
-    LOG.info("Ingestion was successful for topics: " + successTables);
+    log.info("Ingestion was successful for topics: {}", successTables);
     if (!failedTables.isEmpty()) {
-      LOG.info("Ingestion failed for topics: " + failedTables);
+      log.info("Ingestion failed for topics: {}", failedTables);
     }
   }
 
@@ -493,16 +495,4 @@ public static class Constants {
     private static final String UNDERSCORE = "_";
     private static final String COMMA_SEPARATOR = ",";
   }
-
-  public Set<String> getSuccessTables() {
-    return successTables;
-  }
-
-  public Set<String> getFailedTables() {
-    return failedTables;
-  }
-
-  public List<TableExecutionContext> getTableExecutionContexts() {
-    return this.tableExecutionContexts;
-  }
 }