Minor improvements w.r.t #928 & Releasing a newer version (#932)

* Releasing a new version And Minor improvements

* Using immutable empty set & keeping SNAPSHOT to accidently not release any version

---------

Co-authored-by: Shrinand Thakkar <sthakkar@sthakkar-mn2.linkedin.biz>

Author: shrinandthakkar

datastream-server/src/main/java/com/linkedin/datastream/server/EventProducer.java

@@ -122,6 +122,20 @@ public class EventProducer implements DatastreamEventProducer {
    * @param customCheckpointing decides whether producer should use custom checkpointing or the datastream server
    *                            provided checkpointing.
    */
+  public EventProducer(DatastreamTask task, TransportProvider transportProvider, CheckpointProvider checkpointProvider,
+      Properties config, boolean customCheckpointing) {
+    this(task, transportProvider, checkpointProvider, config, customCheckpointing, (t) -> Collections.emptySet());
+  }
+
+  /**
+   * Construct an EventProducer instance.
+   * @param transportProvider the transport provider
+   * @param checkpointProvider the checkpoint provider
+   * @param config the config options
+   * @param customCheckpointing decides whether producer should use custom checkpointing or the datastream server
+   *                            provided checkpointing.
+   * @param throughputViolatingTopicsProvider function parameter per task to find the throughput violating topics
+   */
   public EventProducer(DatastreamTask task, TransportProvider transportProvider, CheckpointProvider checkpointProvider,
       Properties config, boolean customCheckpointing,
       Function<DatastreamTask, Set<String>> throughputViolatingTopicsProvider) {

datastream-server/src/test/java/com/linkedin/datastream/server/TestEventProducer.java

@@ -8,7 +8,6 @@
 import java.lang.reflect.Field;
 import java.util.Collections;
 import java.util.HashMap;
-import java.util.HashSet;
 import java.util.Properties;
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -71,8 +70,8 @@ public void send(String destination, DatastreamProducerRecord record, SendCallba
       }
     };
 
-    EventProducer eventProducer = new EventProducer(task, transport,
-        new NoOpCheckpointProvider(), new Properties(), false, (t) -> new HashSet<>());
+    EventProducer eventProducer =
+        new EventProducer(task, transport, new NoOpCheckpointProvider(), new Properties(), false);
 
     int eventCount = 5;
     for (int i = 0; i < eventCount; i++) {
@@ -122,8 +121,8 @@ public void send(String destination, DatastreamProducerRecord record, SendCallba
       }
     };
 
-    EventProducer eventProducer = new EventProducer(task, transport,
-        new NoOpCheckpointProvider(), new Properties(), false, (t) -> new HashSet<>());
+    EventProducer eventProducer =
+        new EventProducer(task, transport, new NoOpCheckpointProvider(), new Properties(), false);
 
     int eventCount = 5;
     for (int i = 0; i < eventCount; i++) {
@@ -143,8 +142,7 @@ public void testPerDatastreamMetrics() {
 
     Properties props = new Properties();
     props.put(EventProducer.CONFIG_ENABLE_PER_TOPIC_METRICS, Boolean.FALSE.toString());
-    EventProducer eventProducer =
-        new EventProducer(task, transport, new NoOpCheckpointProvider(), props, false, (t) -> new HashSet<>());
+    EventProducer eventProducer = new EventProducer(task, transport, new NoOpCheckpointProvider(), props, false);
 
     eventProducer.send(createDatastreamProducerRecord(), (m, e) -> {
     });

gradle/maven.gradle

@@ -1,5 +1,5 @@
 allprojects {
-  version = "5.2.8-SNAPSHOT"
+  version = "5.3.0-SNAPSHOT"
 }
 
 subprojects {