Disable the replication_latency metric for backfilled items (#146)

Author: aymkhalil

agent/src/main/java/com/datastax/oss/cdc/agent/AbstractPulsarMutationSender.java

@@ -251,14 +251,17 @@ public CompletableFuture<MessageId> sendMutationAsync(final AbstractMutation<T>
             Producer<KeyValue<byte[], MutationValue>> producer = getProducer(mutation);
             SchemaAndWriter schemaAndWriter = getAvroKeySchema(mutation);
             TypedMessageBuilder<KeyValue<byte[], MutationValue>> messageBuilder = producer.newMessage();
-            return messageBuilder
+            messageBuilder = messageBuilder
                     .value(new KeyValue(
                             serializeAvroGenericRecord(buildAvroKey(schemaAndWriter.schema, mutation), schemaAndWriter.writer),
                             mutation.mutationValue()))
-                    .property(Constants.WRITETIME, mutation.getTs() + "")
                     .property(Constants.SEGMENT_AND_POSITION, mutation.getSegment() + ":" + mutation.getPosition())
-                    .property(Constants.TOKEN, mutation.getToken().toString())
-                    .sendAsync();
+                    .property(Constants.TOKEN, mutation.getToken().toString());
+            // a WRITETIME property is only used by the connector to emit e2e latency metric, skip if the mutation is not timestamped
+            if (mutation.getTs() != -1) {
+                messageBuilder = messageBuilder.property(Constants.WRITETIME, mutation.getTs() + "");
+            }
+            return messageBuilder.sendAsync();
         } catch(Exception e) {
             CompletableFuture future = new CompletableFuture<>();
             future.completeExceptionally(e);

backfill-cli/src/main/java/com/datastax/oss/cdc/backfill/importer/PulsarImporter.java

@@ -172,11 +172,9 @@ public ExitStatus importTable() {
                             }
                             return newVal;
                         }).collect(Collectors.toList());
-                        // tsMicro is used to emit e2e metrics by the connectors, if you carry over the C* WRITETIME
-                        // of the source records, the metric will be greatly skewed because those records are historical.
-                        // For now, will mimic the metric by using now()
-                        // TODO: Disable the e2e latency metric if the records are emitted from cdc back-filling CLI
-                        final long tsMicro = Instant.now().toEpochMilli() * 1000;
+                        // Disables the e2e latency metric because the {@link com.datastax.oss.cdc.Constants.WRITETIME}
+                        // property won't be set
+                        final long tsMicro = -1;
                         final AbstractMutation<TableMetadata> mutation =
                                 createMutation(pkValues.toArray(), this.exportedTable.getCassandraTable(), tsMicro);
                         sendMutationAsync(mutation);

backfill-cli/src/test/java/com/datastax/oss/cdc/backfill/PulsarImporterTest.java

@@ -65,6 +65,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Optional;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
@@ -131,6 +132,8 @@ public void testImportPartitionKeyOnly() {
         Mockito.verify(sender, Mockito.times(1)).close();
         List<AbstractMutation<TableMetadata>> pkValues = abstractMutationCaptor.getAllValues();
         assertEquals(2, pkValues.size());
+        assertEquals(-1L, pkValues.get(0).getTs());
+        assertEquals(-1L, pkValues.get(1).getTs());
         List<Object> allPkValues = pkValues.stream().flatMap(v-> Arrays.stream(v.getPkValues())).collect(Collectors.toList());
         assertThat(allPkValues, containsInAnyOrder("id3", "id8"));
     }