Changes per review comments

Added the following test-cases:
- When preserveEventSourceTimestamp is True
- When preserveEventSourceTimestamp is False
- When preserveEventSourceTimestamp is not configured

Author: sanjay24

datastream-kafka/src/main/java/com/linkedin/datastream/kafka/KafkaTransportProvider.java

@@ -82,7 +82,7 @@ public KafkaTransportProvider(DatastreamTask datastreamTask, List<KafkaProducerW
     }
 
     _preserveEventSourceTimestamp = Boolean.parseBoolean(datastreamTask.getDatastreams().get(0).getMetadata()
-            .get(DatastreamMetadataConstants.PRESERVE_EVENT_SOURCE_TIMESTAMP));
+            .getOrDefault(DatastreamMetadataConstants.PRESERVE_EVENT_SOURCE_TIMESTAMP, Boolean.FALSE.toString()));
 
     // initialize metrics
     _dynamicMetricsManager = DynamicMetricsManager.getInstance();
@@ -101,7 +101,7 @@ private ProducerRecord<byte[], byte[]> convertToProducerRecord(String topicName,
 
     Optional<Integer> partition = record.getPartition();
 
-    byte[] keyValue = new byte[0];
+    byte[] keyValue = null;
     byte[] payloadValue = new byte[0];
     if (event instanceof BrooklinEnvelope) {
       BrooklinEnvelope envelope = (BrooklinEnvelope) event;
@@ -125,7 +125,7 @@ private ProducerRecord<byte[], byte[]> convertToProducerRecord(String topicName,
       // If the partition is not specified. We use the partitionKey as the key. Kafka will use the hash of that
       // to determine the partition. If partitionKey does not exist, use the key value.
       keyValue = record.getPartitionKey().isPresent()
-          ? record.getPartitionKey().get().getBytes(StandardCharsets.UTF_8) : null;
+              ? record.getPartitionKey().get().getBytes(StandardCharsets.UTF_8) : keyValue;
       return new ProducerRecord<>(topicName, null, recordTimeStamp, keyValue, payloadValue);
     }
   }

datastream-kafka/src/test/java/com/linkedin/datastream/kafka/TestKafkaTransportProvider.java

@@ -209,10 +209,20 @@ public void testEventWithoutKeyValueAndPartition() throws Exception {
   }
 
   @Test
-  public void testEventWithTimestamp() throws Exception {
+  public void testEventWithPreserveTimestampTrue() throws Exception {
     testEventSendWithTimestamp(1, 2, -1, false, false, "test", true);
   }
 
+  @Test
+  public void testEventWithPreserveTimestampFalse() throws Exception {
+    testEventSendWithTimestamp(1, 2, -1, false, false, "test", false);
+  }
+
+  @Test
+  public void testEventWithoutPreservingTimestamp() throws Exception {
+    testEventSendWithTimestamp(1, 2, -1, false, false, "test", null);
+  }
+
   @Test
   public void testSendMultipleEventsInSingleDatastreamProducerRecord() throws Exception {
     String metricsPrefix = "test";
@@ -290,7 +300,7 @@ public void testSendMultipleEventsInSingleDatastreamProducerRecord() throws Exce
   }
 
   private void testEventSendWithTimestamp(int numberOfEvents, int numberOfPartitions, int partition, boolean includeKey,
-                             boolean includeValue, String metricsPrefix, boolean preserveSourceEventTimestamp) throws Exception {
+                             boolean includeValue, String metricsPrefix, Boolean preserveSourceEventTimestamp) throws Exception {
     String topicName = getUniqueTopicName();
 
     if (metricsPrefix != null) {
@@ -302,7 +312,9 @@ private void testEventSendWithTimestamp(int numberOfEvents, int numberOfPartitio
 
     Datastream ds = DatastreamTestUtils.createDatastream("test", "ds1", "source", destinationUri, numberOfPartitions);
 
-    ds.getMetadata().put(DatastreamMetadataConstants.PRESERVE_EVENT_SOURCE_TIMESTAMP, Boolean.TRUE.toString());
+    if (preserveSourceEventTimestamp != null) {
+      ds.getMetadata().put(DatastreamMetadataConstants.PRESERVE_EVENT_SOURCE_TIMESTAMP, preserveSourceEventTimestamp.toString());
+    }
 
     DatastreamTask task = new DatastreamTaskImpl(Collections.singletonList(ds));
     TransportProvider transportProvider = provider.assignTransportProvider(task);
@@ -322,7 +334,7 @@ private void testEventSendWithTimestamp(int numberOfEvents, int numberOfPartitio
     }
 
     List<DatastreamProducerRecord> datastreamEvents =
-          createEvents(topicName, partition, numberOfEvents, includeKey, includeValue, eventSourceTimestamps);
+            createEvents(topicName, partition, numberOfEvents, includeKey, includeValue, eventSourceTimestamps);
 
     LOG.info(String.format("Trying to send %d events to topic %s", datastreamEvents.size(), topicName));
 
@@ -343,7 +355,11 @@ private void testEventSendWithTimestamp(int numberOfEvents, int numberOfPartitio
       return readTimestamps.size() < numberOfEvents;
     });
 
-    Assert.assertEquals(readTimestamps, eventSourceTimestamps);
+    if (preserveSourceEventTimestamp != null && preserveSourceEventTimestamp) {
+      Assert.assertEquals(readTimestamps, eventSourceTimestamps);
+    } else {
+      Assert.assertNotEquals(readTimestamps, eventSourceTimestamps);
+    }
 
     if (metricsPrefix != null) {
       // verify that configured metrics prefix was used