MET-6144 Review remarks - part 3 Code formating

Author: marcin-rp

commons/src/main/java/eu/europeana/processing/model/DataPartition.java

@@ -8,25 +8,24 @@
 
 /**
  * Class describing actual partition of data that is delivered to reader
- *
  */
 @Value
 @AllArgsConstructor
 public class DataPartition implements AbstractPartition {
 
-    @Serial
-    private static final long serialVersionUID = 1;
+  @Serial
+  private static final long serialVersionUID = 1;
 
-    long offset;
-    long limit;
-    @With
-    long progress;
-    @With
-    UUID enumeratorId;
+  long offset;
+  long limit;
+  @With
+  long progress;
+  @With
+  UUID enumeratorId;
 
-    @Override
-    public String splitId() {
-        return offset+"_"+limit;
-    }
+  @Override
+  public String splitId() {
+    return offset + "_" + limit;
+  }
 
 }

commons/src/main/java/eu/europeana/processing/source/AbstractDbEnumerator.java

@@ -9,9 +9,11 @@
 
 /**
  * Flink enumerator that provides splits for Metis jobs using as a source PostgresDB
+ *
  * @param <S> state used by an implementation of enumerator which is saved in snapshot
  */
-public abstract class AbstractDbEnumerator<S extends DbEnumeratorState> extends AbstractEnumerator<DataPartition,S> {
+public abstract class AbstractDbEnumerator<S extends DbEnumeratorState> extends AbstractEnumerator<DataPartition, S> {
+
   private static final Logger LOGGER = LoggerFactory.getLogger(AbstractDbEnumerator.class);
 
 
@@ -67,7 +69,7 @@ protected DataPartition createNextPartition() {
 
   @Override
   public S snapshotState(long checkpointId) {
-    S state=super.snapshotState(checkpointId);
+    S state = super.snapshotState(checkpointId);
     state.setRecordsToBeProcessed(recordsToBeProcessed);
     return state;
   }

commons/src/main/java/eu/europeana/processing/source/AbstractEnumerator.java

@@ -25,10 +25,12 @@
 
 /**
  * Base Flink enumerator base for different sources, implementing checkpointing failover.
+ *
  * @param <P> split type that is used by the source implementation
  * @param <S> state used by an implementation of enumerator which is saved in snapshot
  */
-public abstract class AbstractEnumerator<P extends AbstractPartition,S extends AbstractEnumeratorState<P>> implements SplitEnumerator<P, S> {
+public abstract class AbstractEnumerator<P extends AbstractPartition, S extends AbstractEnumeratorState<P>> implements
+    SplitEnumerator<P, S> {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(AbstractEnumerator.class);
   private static final int DEFAULT_CHUNK_SIZE = 1000;
@@ -142,14 +144,14 @@ private void addSplitBack(P split, int subtaskId) {
     if (inExecutingMap != null) {
       split = updateProgress(inExecutingMap, split.getProgress());
     }
-    returnedPartitions.put(split.splitId(), updateProgress(returnedPartitions.get(split.splitId()),split.getProgress()));
+    returnedPartitions.put(split.splitId(), updateProgress(returnedPartitions.get(split.splitId()), split.getProgress()));
     LOGGER.info(
         "Added split: {} from subtask: {} back. Currently executing: {} splits, all returned splits: {}",
         split, subtaskId, executingPartitions.size(), returnedPartitions.size());
   }
 
   private P updateProgress(P previousSplit, long progress) {
-    LOGGER.debug("Updating progress to: {}, for split: {}",progress, previousSplit);
+    LOGGER.debug("Updating progress to: {}, for split: {}", progress, previousSplit);
     P current = (P) previousSplit.withProgress(progress);
     long progressIncrease = current.getProgress() - previousSplit.getProgress();
     emittedRecordCount += progressIncrease;
@@ -234,7 +236,6 @@ public void close() throws IOException {
   }
 
 
-
   private List<P> getIncompletePartitionsSnapshot() {
     List<P> incompletePartitions = new ArrayList<>();
     incompletePartitions.addAll(executingPartitions.values());

commons/src/main/java/eu/europeana/processing/source/DbEnumeratorState.java

@@ -20,6 +20,7 @@
 @NoArgsConstructor
 @SuperBuilder
 public class DbEnumeratorState extends AbstractEnumeratorState<DataPartition> {
+
   @Serial
   private static final long serialVersionUID = 2;
 }

commons/src/main/java/eu/europeana/processing/source/DbSourceWithProgressHandling.java

@@ -14,45 +14,45 @@
 
 public class DbSourceWithProgressHandling implements Source<ExecutionRecord, DataPartition, DbEnumeratorState> {
 
-    @Serial
-    private static final long serialVersionUID = 1;
-
-    private final ParameterTool parameterTool;
-
-    public DbSourceWithProgressHandling(ParameterTool parameterTool) {
-        this.parameterTool = parameterTool;
-    }
-
-    @Override
-    public Boundedness getBoundedness() {
-        return Boundedness.BOUNDED;
-    }
-
-    @Override
-    public SplitEnumerator<DataPartition, DbEnumeratorState> createEnumerator(SplitEnumeratorContext<DataPartition> enumContext) {
-        return new RegularDbEnumerator(enumContext, parameterTool);
-    }
-
-    @Override
-    public SplitEnumerator<DataPartition, DbEnumeratorState> restoreEnumerator(
-        SplitEnumeratorContext<DataPartition> enumContext,
-        DbEnumeratorState state) {
-        return new RegularDbEnumerator(enumContext, parameterTool, state);
-    }
-
-    @Override
-    public SourceReader<ExecutionRecord, DataPartition> createReader(SourceReaderContext readerContext) {
-        return new RegularDbReader(readerContext, parameterTool);
-    }
-
-    @Override
-    public SimpleVersionedSerializer<DataPartition> getSplitSerializer() {
-        return new ObjectStreamVersionedSerializer<>();
-    }
-
-    @Override
-    public SimpleVersionedSerializer<DbEnumeratorState> getEnumeratorCheckpointSerializer() {
-        return new ObjectStreamVersionedSerializer<>();
-    }
+  @Serial
+  private static final long serialVersionUID = 1;
+
+  private final ParameterTool parameterTool;
+
+  public DbSourceWithProgressHandling(ParameterTool parameterTool) {
+    this.parameterTool = parameterTool;
+  }
+
+  @Override
+  public Boundedness getBoundedness() {
+    return Boundedness.BOUNDED;
+  }
+
+  @Override
+  public SplitEnumerator<DataPartition, DbEnumeratorState> createEnumerator(SplitEnumeratorContext<DataPartition> enumContext) {
+    return new RegularDbEnumerator(enumContext, parameterTool);
+  }
+
+  @Override
+  public SplitEnumerator<DataPartition, DbEnumeratorState> restoreEnumerator(
+      SplitEnumeratorContext<DataPartition> enumContext,
+      DbEnumeratorState state) {
+    return new RegularDbEnumerator(enumContext, parameterTool, state);
+  }
+
+  @Override
+  public SourceReader<ExecutionRecord, DataPartition> createReader(SourceReaderContext readerContext) {
+    return new RegularDbReader(readerContext, parameterTool);
+  }
+
+  @Override
+  public SimpleVersionedSerializer<DataPartition> getSplitSerializer() {
+    return new ObjectStreamVersionedSerializer<>();
+  }
+
+  @Override
+  public SimpleVersionedSerializer<DbEnumeratorState> getEnumeratorCheckpointSerializer() {
+    return new ObjectStreamVersionedSerializer<>();
+  }
 
 }

commons/src/main/java/eu/europeana/processing/source/ProgressSnapshotEvent.java

@@ -6,9 +6,9 @@
 import org.apache.flink.api.connector.source.SourceEvent;
 
 /**
- * Event with progress snapshot was created on a reader - number of record emitted by reader to execution.
- * The event does not mean that records are already stored in DB, but contains checkpointId, so could be
- * held and used when given checkpoint is completed.
+ * Event with progress snapshot was created on a reader - number of record emitted by reader to execution. The event does not mean
+ * that records are already stored in DB, but contains checkpointId, so could be held and used when given checkpoint is
+ * completed.
  */
 @Value
 public class ProgressSnapshotEvent implements SourceEvent {

commons/src/main/java/eu/europeana/processing/source/ProgressUpdater.java

@@ -24,11 +24,11 @@ public class ProgressUpdater implements Closeable {
 
   /**
    * Creates ProgressUpdater
+   *
    * @param parameterTool - all the command line parameters of the job
-   * @param completedFilesCount - Number of files already completed. It is greater than 0 only if the
-   * source is restored from a checkpoint.
+   * @param completedFilesCount - Number of files already completed. It is greater than 0 only if the source is restored from a
+   * checkpoint.
    * @param dbConnectionProvider - db connection provider
-   *
    */
   public ProgressUpdater(DbConnectionProvider dbConnectionProvider, ParameterTool parameterTool, long completedFilesCount) {
     this.dbConnectionProvider = dbConnectionProvider;
@@ -40,8 +40,9 @@ public ProgressUpdater(DbConnectionProvider dbConnectionProvider, ParameterTool
 
 
   /**
-   * Stores count of emitted files. It is invoked during doing snapshot of enumerator state.
-   * This count that is later saved as progress into DB after the checkpoint is completed.
+   * Stores count of emitted files. It is invoked during doing snapshot of enumerator state. This count that is later saved as
+   * progress into DB after the checkpoint is completed.
+   *
    * @param shapshotedEmittedFilesCount - number of emitted files.
    */
   public void snapshotEmittedFilesCount(long shapshotedEmittedFilesCount) {
@@ -50,8 +51,8 @@ public void snapshotEmittedFilesCount(long shapshotedEmittedFilesCount) {
   }
 
   /**
-   * Saves previously stored count of emitted files in the DB. Invoked after the checkpoint is completed,
-   * what means that the emitted files are already stored in the DB.
+   * Saves previously stored count of emitted files in the DB. Invoked after the checkpoint is completed, what means that the
+   * emitted files are already stored in the DB.
    */
   public void saveProgressInDB() {
     if (snapshottedEmittedFilesCount != lastStoredFilesCount) {
@@ -63,14 +64,14 @@ public void saveProgressInDB() {
       taskInfoRepo.update(taskInfo);
       lastStoredFilesCount = snapshottedEmittedFilesCount;
       LOGGER.info("Updated task progress in DB: {}", taskInfo);
-    }else{
+    } else {
       LOGGER.info("Need not update progress for: {}", taskId);
     }
   }
 
   public void close() {
-      dbConnectionProvider.close();
-      LOGGER.debug("Closed: {}", ProgressUpdater.class.getSimpleName());
+    dbConnectionProvider.close();
+    LOGGER.debug("Closed: {}", ProgressUpdater.class.getSimpleName());
   }
 
 }

commons/src/main/java/eu/europeana/processing/source/RegularDbEnumerator.java

@@ -34,7 +34,7 @@ public RegularDbEnumerator(SplitEnumeratorContext<DataPartition> context, Parame
    * @param state enumerator state container
    */
   public RegularDbEnumerator(SplitEnumeratorContext<DataPartition> context,
-      ParameterTool parameterTool,DbEnumeratorState state) {
+      ParameterTool parameterTool, DbEnumeratorState state) {
     super(context, parameterTool, state);
   }
 

commons/src/main/java/eu/europeana/processing/source/RegularDbReader.java

@@ -15,11 +15,13 @@
  * SourceReader implementation for regular - all but not harvesting jobs.
  */
 public class RegularDbReader extends AbstractDbReader<ExecutionRecord> {
+
   private static final Logger LOGGER = LoggerFactory.getLogger(RegularDbReader.class);
   private ExecutionRecordRepository executionRecordRepository;
 
   /**
    * Creates RegularDbReader
+   *
    * @param context - Flink context
    * @param parameterTool - job parameters
    */

http/src/main/java/eu/europeana/processing/http/reader/HttpEnumerator.java

@@ -17,8 +17,8 @@
 
 /**
  * The SplitEnumerator implementation for HttpSource. It downloads an archive file and provided splits containing lists of the
- * files from the archive and chosen extraction mode - depending on the archive type. Depending on the archive type, the archive also
- * is or is not extracted to the shared temporary directory - more described in: {@link ArchiveFileNamesExtractor}
+ * files from the archive and chosen extraction mode - depending on the archive type. Depending on the archive type, the archive
+ * also is or is not extracted to the shared temporary directory - more described in: {@link ArchiveFileNamesExtractor}
  */
 public class HttpEnumerator extends AbstractEnumerator<HttpSourceSplit, HttpEnumeratorState> {
 
@@ -123,7 +123,7 @@ protected HttpSourceSplit createNextPartition() {
 
       startedRecordsCount += fileNames.size();
       return split;
-    }else{
+    } else {
       return null;
     }
   }