refactor: apply some Java best practices

src/main/java/io/numaproj/numaflow/accumulator/Constants.java

@@ -1,19 +1,18 @@
 package io.numaproj.numaflow.accumulator;
 
 class Constants {
-    public static final String SUCCESS = "SUCCESS";
-
-    public static final String DELIMITER = ":";
-
-    public static final int DEFAULT_MESSAGE_SIZE = 1024 * 1024 * 64;
-
-    public static final String DEFAULT_SOCKET_PATH = "/var/run/numaflow/accumulator.sock";
-
-    public static final String DEFAULT_SERVER_INFO_FILE_PATH = "/var/run/numaflow/accumulator-server-info";
-
-    public static final int DEFAULT_PORT = 50051;
-
-    public static final String DEFAULT_HOST = "localhost";
-
-    public static final String EOF = "EOF";
+  public static final String SUCCESS = "SUCCESS";
+  public static final String DELIMITER = ":";
+  public static final int DEFAULT_MESSAGE_SIZE = 1024 * 1024 * 64;
+  public static final String DEFAULT_SOCKET_PATH = "/var/run/numaflow/accumulator.sock";
+  public static final String DEFAULT_SERVER_INFO_FILE_PATH =
+      "/var/run/numaflow/accumulator-server-info";
+  public static final int DEFAULT_PORT = 50051;
+  public static final String DEFAULT_HOST = "localhost";
+  public static final String EOF = "EOF";
+
+  // Private constructor to prevent instantiation
+  private Constants() {
+    throw new IllegalStateException("Utility class 'Constants' should not be instantiated");
+  }
 }

src/main/java/io/numaproj/numaflow/accumulator/model/Message.java

@@ -1,65 +1,61 @@
 package io.numaproj.numaflow.accumulator.model;
 
-import lombok.Getter;
-
 import java.time.Instant;
 import java.util.Map;
+import lombok.Getter;
 
-/**
- * Message is used to wrap the data returned by Accumulator functions.
- */
+/** Message is used to wrap the data returned by Accumulator functions. */
 @Getter
 public class Message {
-    private final Instant eventTime;
-    private final Instant watermark;
-    private final Map<String, String> headers;
-    private final String id;
-    private String[] keys;
-    private byte[] value;
-    private String[] tags;
+  private final Instant eventTime;
+  private final Instant watermark;
+  private final Map<String, String> headers;
+  private final String id;
+  private String[] keys;
+  private byte[] value;
+  private String[] tags;
 
-    /**
-     * Constructor for constructing message from Datum, it is advised to use the
-     * incoming Datum to construct the message, because event time, watermark, id
-     * and headers of the message are derived from the Datum. Only use custom
-     * implementation of the Datum if you know what you are doing.
-     *
-     * @param datum {@link Datum} object
-     */
-    public Message(Datum datum) {
-        this.keys = datum.getKeys();
-        this.value = datum.getValue();
-        this.headers = datum.getHeaders();
-        this.eventTime = datum.getEventTime();
-        this.watermark = datum.getWatermark();
-        this.id = datum.getID();
-        this.tags = null;
-    }
+  /**
+   * Constructor for constructing message from Datum, it is advised to use the incoming Datum to
+   * construct the message, because event time, watermark, id and headers of the message are derived
+   * from the Datum. Only use custom implementation of the Datum if you know what you are doing.
+   *
+   * @param datum {@link Datum} object
+   */
+  public Message(Datum datum) {
+    this.keys = datum.getKeys();
+    this.value = datum.getValue();
+    this.headers = datum.getHeaders();
+    this.eventTime = datum.getEventTime();
+    this.watermark = datum.getWatermark();
+    this.id = datum.getID();
+    this.tags = null;
+  }
 
-    /*
-     * sets the value of the message
-     *
-     * @param value byte array of the value
-     */
-    public void setValue(byte[] value) {
-        this.value = value;
-    }
+  /*
+   * sets the value of the message
+   *
+   * @param value byte array of the value
+   */
+  public void setValue(byte[] value) {
+    this.value = value;
+  }
 
-    /*
-     * sets the keys of the message
-     *
-     * @param keys string array of the keys
-     */
-    public void setKeys(String[] keys) {
-        this.keys = keys;
-    }
+  /*
+   * sets the keys of the message
+   *
+   * @param keys string array of the keys
+   */
+  public void setKeys(String[] keys) {
+    this.keys = keys;
+  }
 
-    /*
-     * sets the tags of the message, tags are used for conditional forwarding
-     *
-     * @param tags string array of the tags
-     */
-    public void setTags(String[] tags) {
-        this.tags = tags;
-    }
+  /*
+   * sets the tags of the message, tags are used for conditional forwarding
+   *
+   * @param tags string array of the tags
+   */
+  public void setTags(String[] tags) {
+    this.tags = tags;
+  }
 }

src/main/java/io/numaproj/numaflow/batchmapper/Constants.java

@@ -1,20 +1,17 @@
 package io.numaproj.numaflow.batchmapper;
 
 class Constants {
-    public static final int DEFAULT_MESSAGE_SIZE = 1024 * 1024 * 64;
-
-    public static final String DEFAULT_SOCKET_PATH = "/var/run/numaflow/batchmap.sock";
-
-    public static final String DEFAULT_SERVER_INFO_FILE_PATH = "/var/run/numaflow/mapper-server-info";
-
-    public static final int DEFAULT_PORT = 50051;
-
-    public static final String DEFAULT_HOST = "localhost";
-
-    public static final String SUCCESS = "SUCCESS";
-
-    public static final String MAP_MODE_KEY = "MAP_MODE";
-
-    public static final String MAP_MODE = "batch-map";
+  public static final int DEFAULT_MESSAGE_SIZE = 1024 * 1024 * 64;
+  public static final String DEFAULT_SOCKET_PATH = "/var/run/numaflow/batchmap.sock";
+  public static final String DEFAULT_SERVER_INFO_FILE_PATH = "/var/run/numaflow/mapper-server-info";
+  public static final int DEFAULT_PORT = 50051;
+  public static final String DEFAULT_HOST = "localhost";
+  public static final String SUCCESS = "SUCCESS";
+  public static final String MAP_MODE_KEY = "MAP_MODE";
+  public static final String MAP_MODE = "batch-map";
+
+  // Private constructor to prevent instantiation
+  private Constants() {
+    throw new IllegalStateException("Utility class 'Constants' should not be instantiated");
+  }
 }
-

src/main/java/io/numaproj/numaflow/batchmapper/Message.java

@@ -2,57 +2,53 @@
 
 import lombok.Getter;
 
-/**
- * Message is used to wrap the data returned by Mapper.
- */
-
+/** Message is used to wrap the data returned by Mapper. */
 @Getter
 public class Message {
-    public static final String DROP = "U+005C__DROP__";
-
-    private final String[] keys;
-    private final byte[] value;
-    private final String[] tags;
-
-
-    /**
-     * used to create Message with value, keys and tags(used for conditional forwarding)
-     *
-     * @param value message value
-     * @param keys message keys
-     * @param tags message tags which will be used for conditional forwarding
-     */
-    public Message(byte[] value, String[] keys, String[] tags) {
-        this.keys = keys;
-        this.value = value;
-        this.tags = tags;
-    }
-
-    /**
-     * used to create Message with value.
-     *
-     * @param value message value
-     */
-    public Message(byte[] value) {
-        this(value, null, null);
-    }
-
-    /**
-     * used to create Message with value and keys.
-     *
-     * @param value message value
-     * @param keys message keys
-     */
-    public Message(byte[] value, String[] keys) {
-        this(value, keys, null);
-    }
-
-    /**
-     * creates a Message which will be dropped
-     *
-     * @return returns the Message which will be dropped
-     */
-    public static Message toDrop() {
-        return new Message(new byte[0], null, new String[]{DROP});
-    }
+  private static final String[] DROP_TAGS = {"U+005C__DROP__"};
+  private final String[] keys;
+  private final byte[] value;
+  private final String[] tags;
+
+  /**
+   * used to create Message with value, keys and tags(used for conditional forwarding)
+   *
+   * @param value message value
+   * @param keys message keys
+   * @param tags message tags which will be used for conditional forwarding
+   */
+  public Message(byte[] value, String[] keys, String[] tags) {
+    // defensive copy - once the Message is created, the caller should not be able to modify it.
+    this.keys = keys == null ? null : keys.clone();
+    this.value = value == null ? null : value.clone();
+    this.tags = tags == null ? null : tags.clone();
+  }
+
+  /**
+   * used to create Message with value.
+   *
+   * @param value message value
+   */
+  public Message(byte[] value) {
+    this(value, null, null);
+  }
+
+  /**
+   * used to create Message with value and keys.
+   *
+   * @param value message value
+   * @param keys message keys
+   */
+  public Message(byte[] value, String[] keys) {
+    this(value, keys, null);
+  }
+
+  /**
+   * creates a Message which will be dropped
+   *
+   * @return returns the Message which will be dropped
+   */
+  public static Message toDrop() {
+    return new Message(new byte[0], null, DROP_TAGS);
+  }
 }

src/main/java/io/numaproj/numaflow/info/ServerInfo.java

@@ -1,15 +1,14 @@
 package io.numaproj.numaflow.info;
 
+import static java.util.Map.entry;
+
 import com.fasterxml.jackson.annotation.JsonProperty;
+import java.util.Map;
 import lombok.AllArgsConstructor;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
 import lombok.Setter;
 
-import java.util.Map;
-
-import static java.util.Map.entry;
-
 /**
  * Server Information to be used by client to determine:
  * - protocol: what is right protocol to use (UDS or TCP)
@@ -22,7 +21,7 @@
 @Setter
 @NoArgsConstructor
 @AllArgsConstructor
-public class ServerInfo {
+public final class ServerInfo {
     // Specify the minimum Numaflow version required by the current SDK version
     // To update this value, please follow the instructions for
     // MINIMUM_NUMAFLOW_VERSION in

src/main/java/io/numaproj/numaflow/info/ServerInfoAccessor.java

@@ -1,8 +1,6 @@
 package io.numaproj.numaflow.info;
 
 public interface ServerInfoAccessor {
-    String INFO_EOF = "U+005C__END__";
-
     /**
      * Get current runtime numaflow-java SDK version.
      */
@@ -11,7 +9,6 @@ public interface ServerInfoAccessor {
     /**
      * Delete filePath if it exists.
      * Write serverInfo to filePath in Json format.
-     * Append {@link ServerInfoAccessor#INFO_EOF} as a new line to indicate end of file.
      *
      * @param serverInfo server information POJO
      * @param filePath file path to write to