You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by ca...@apache.org on 2019/07/29 17:19:44 UTC
[samza] branch master updated: Revert "Fixing SamzaSQL Latency
Metrics resolution (#1116)" (#1118)
This is an automated email from the ASF dual-hosted git repository.
cameronlee pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/samza.git
The following commit(s) were added to refs/heads/master by this push:
new 97942d9 Revert "Fixing SamzaSQL Latency Metrics resolution (#1116)" (#1118)
97942d9 is described below
commit 97942d9ae1a63c420c64beb84b2a923e6a96b450
Author: Cameron Lee <ca...@linkedin.com>
AuthorDate: Mon Jul 29 10:19:39 2019 -0700
Revert "Fixing SamzaSQL Latency Metrics resolution (#1116)" (#1118)
This reverts commit ec536e14a1aa3d0fbd31c76418b605ea82c8b8d9.
PR 1116 included some issues, which will be fixed in PR 1117: 1) semantic change to return values of IncomingMessageEnvelope.getEventTime and IncomingMessageEnvelope.getArrivalTime (changed from millis to nanos) which are public API methods, and it will be hard to find existing usages to make sure they are updated properly, 2) usages of constructor of IncomingMessageEnvelope was not updated to handle nanos correctly in certain places (e.g. KafkaConsumerProxy)
---
.../samza/system/IncomingMessageEnvelope.java | 8 +++---
.../apache/samza/sql/SamzaSqlInputTransformer.java | 4 ++-
.../apache/samza/sql/avro/AvroRelConverter.java | 4 +--
.../apache/samza/sql/data/SamzaSqlRelMessage.java | 6 ++--
.../samza/sql/data/SamzaSqlRelMsgMetadata.java | 30 ++++++++++----------
.../samza/sql/translator/FilterTranslator.java | 16 ++++++-----
.../sql/translator/LogicalAggregateTranslator.java | 3 +-
.../samza/sql/translator/ProjectTranslator.java | 15 +++++-----
.../samza/sql/translator/QueryTranslator.java | 31 ++++++++++----------
.../samza/sql/translator/ScanTranslator.java | 17 +++++------
.../samza/sql/translator/TranslatorConstants.java | 8 +++---
.../org/apache/samza/sql/util/TranslatorUtils.java | 33 ----------------------
.../samza/sql/data/TestSamzaSqlRelMessage.java | 17 ++++++-----
.../serializers/TestSamzaSqlRelMessageSerde.java | 3 +-
.../serializers/TestSamzaSqlRelRecordSerde.java | 2 +-
.../samza/sql/translator/TestFilterTranslator.java | 2 +-
.../sql/translator/TestProjectTranslator.java | 2 +-
.../TestSamzaSqlLocalTableJoinFunction.java | 16 ++++-------
.../TestSamzaSqlRemoteTableJoinFunction.java | 2 +-
19 files changed, 94 insertions(+), 125 deletions(-)
diff --git a/samza-api/src/main/java/org/apache/samza/system/IncomingMessageEnvelope.java b/samza-api/src/main/java/org/apache/samza/system/IncomingMessageEnvelope.java
index e8bb331..7392253 100644
--- a/samza-api/src/main/java/org/apache/samza/system/IncomingMessageEnvelope.java
+++ b/samza-api/src/main/java/org/apache/samza/system/IncomingMessageEnvelope.java
@@ -37,10 +37,10 @@ public class IncomingMessageEnvelope {
private final Object key;
private final Object message;
private final int size;
- // the System's nanoTime when this event occurred, should be set by the event source, 0 means unassigned
+ // the timestamp when this event occured, should be set by the event source, 0 means unassigned
private long eventTime = 0L;
- // the System's nanoTime when this event is picked-up by samza, 0 means unassgined
- private long arrivalTime;
+ // the timestamp when this event is pickedup by samza, 0 means unassgined
+ private long arrivalTime = 0L;
/**
* Constructs a new IncomingMessageEnvelope from specified components.
@@ -70,7 +70,7 @@ public class IncomingMessageEnvelope {
this.key = key;
this.message = message;
this.size = size;
- this.arrivalTime = System.nanoTime();
+ this.arrivalTime = Instant.now().toEpochMilli();
}
/**
diff --git a/samza-sql/src/main/java/org/apache/samza/sql/SamzaSqlInputTransformer.java b/samza-sql/src/main/java/org/apache/samza/sql/SamzaSqlInputTransformer.java
index 6bcae8b..2bbce22 100644
--- a/samza-sql/src/main/java/org/apache/samza/sql/SamzaSqlInputTransformer.java
+++ b/samza-sql/src/main/java/org/apache/samza/sql/SamzaSqlInputTransformer.java
@@ -40,7 +40,9 @@ public class SamzaSqlInputTransformer implements InputTransformer {
public Object apply(IncomingMessageEnvelope ime) {
Assert.notNull(ime, "ime is null");
KV<Object, Object> keyAndMessageKV = KV.of(ime.getKey(), ime.getMessage());
- SamzaSqlRelMsgMetadata metadata = new SamzaSqlRelMsgMetadata(ime.getEventTime(), ime.getArrivalTime(),0L);
+ SamzaSqlRelMsgMetadata metadata = new SamzaSqlRelMsgMetadata(
+ (ime.getEventTime() == 0) ? "" : Instant.ofEpochMilli(ime.getEventTime()).toString(),
+ (ime.getArrivalTime() == 0) ? "" : Instant.ofEpochMilli(ime.getArrivalTime()).toString(), null);
SamzaSqlInputMessage samzaMsg = SamzaSqlInputMessage.of(keyAndMessageKV, metadata);
return samzaMsg;
}
diff --git a/samza-sql/src/main/java/org/apache/samza/sql/avro/AvroRelConverter.java b/samza-sql/src/main/java/org/apache/samza/sql/avro/AvroRelConverter.java
index 3d3ebe2..fcb0c72 100644
--- a/samza-sql/src/main/java/org/apache/samza/sql/avro/AvroRelConverter.java
+++ b/samza-sql/src/main/java/org/apache/samza/sql/avro/AvroRelConverter.java
@@ -96,7 +96,7 @@ public class AvroRelConverter implements SamzaRelConverter {
}
return new SamzaSqlRelMessage(samzaMessage.getKey(), payloadFieldNames, payloadFieldValues,
- new SamzaSqlRelMsgMetadata(0, 0, 0));
+ new SamzaSqlRelMsgMetadata("", "", ""));
}
/**
@@ -108,7 +108,7 @@ public class AvroRelConverter implements SamzaRelConverter {
List<Object> payloadFieldValues = new ArrayList<>();
fetchFieldNamesAndValuesFromIndexedRecord(record, payloadFieldNames, payloadFieldValues, schema);
return new SamzaSqlRelMessage(key, payloadFieldNames, payloadFieldValues,
- new SamzaSqlRelMsgMetadata(0, 0, 0));
+ new SamzaSqlRelMsgMetadata("", "", ""));
}
public static void fetchFieldNamesAndValuesFromIndexedRecord(IndexedRecord record, List<String> fieldNames,
diff --git a/samza-sql/src/main/java/org/apache/samza/sql/data/SamzaSqlRelMessage.java b/samza-sql/src/main/java/org/apache/samza/sql/data/SamzaSqlRelMessage.java
index 84db61f..d538a8e 100644
--- a/samza-sql/src/main/java/org/apache/samza/sql/data/SamzaSqlRelMessage.java
+++ b/samza-sql/src/main/java/org/apache/samza/sql/data/SamzaSqlRelMessage.java
@@ -130,15 +130,15 @@ public class SamzaSqlRelMessage implements Serializable {
return key;
}
- public void setEventTime(long eventTime) {
+ public void setEventTime(String eventTime) {
this.samzaSqlRelMsgMetadata.setEventTime(eventTime);
}
- public void setArrivalTime(long arrivalTime) {
+ public void setArrivalTime(String arrivalTime) {
this.samzaSqlRelMsgMetadata.setArrivalTime(arrivalTime);
}
- public void setScanTime(long scanTime) {
+ public void setScanTime(String scanTime) {
this.samzaSqlRelMsgMetadata.setScanTime(scanTime);
}
diff --git a/samza-sql/src/main/java/org/apache/samza/sql/data/SamzaSqlRelMsgMetadata.java b/samza-sql/src/main/java/org/apache/samza/sql/data/SamzaSqlRelMsgMetadata.java
index 60a4b11..14f2892 100644
--- a/samza-sql/src/main/java/org/apache/samza/sql/data/SamzaSqlRelMsgMetadata.java
+++ b/samza-sql/src/main/java/org/apache/samza/sql/data/SamzaSqlRelMsgMetadata.java
@@ -56,60 +56,60 @@ public class SamzaSqlRelMsgMetadata implements Serializable {
* TODO: copy eventTime through from source to RelMessage
*/
@JsonProperty("eventTime")
- private long eventTime;
+ private String eventTime;
/**
* the timestamp of when Samza App received the event
* TODO: set arrivalTime during conversion from IME to SamzaMessage
*/
@JsonProperty("arrivalTime")
- private long arrivalTime;
+ private String arrivalTime;
/**
* the timestamp when SamzaSQL query starts processing the event
* set by the SamzaSQL Scan operator
*/
@JsonProperty("scanTime")
- private long scanTime;
+ private String scanTime;
- public SamzaSqlRelMsgMetadata(@JsonProperty("eventTime") long eventTime, @JsonProperty("arrivalTime") long arrivalTime,
- @JsonProperty("scanTime") long scanTime) {
+ public SamzaSqlRelMsgMetadata(@JsonProperty("eventTime") String eventTime, @JsonProperty("arrivalTime") String arrivalTime,
+ @JsonProperty("scanTime") String scanTime) {
this.eventTime = eventTime;
this.arrivalTime = arrivalTime;
this.scanTime = scanTime;
}
- public SamzaSqlRelMsgMetadata(long eventTime, long arrivalTime, long scanTime, boolean isNewInputMessage) {
+ public SamzaSqlRelMsgMetadata(String eventTime, String arrivalTime, String scanTime, boolean isNewInputMessage) {
this(eventTime, arrivalTime, scanTime);
this.isNewInputMessage = isNewInputMessage;
}
@JsonProperty("eventTime")
- public long getEventTime() { return eventTime;}
+ public String getEventTime() { return eventTime;}
- public void setEventTime(long eventTime) {
+ public void setEventTime(String eventTime) {
this.eventTime = eventTime;
}
- public boolean hasEventTime() { return eventTime != 0; }
+ public boolean hasEventTime() { return eventTime != null && !eventTime.isEmpty(); }
@JsonProperty("arrivalTime")
- public long getarrivalTime() { return arrivalTime;}
+ public String getarrivalTime() { return arrivalTime;}
- public void setArrivalTime(long arrivalTime) {
+ public void setArrivalTime(String arrivalTime) {
this.arrivalTime = arrivalTime;
}
- public boolean hasArrivalTime() { return arrivalTime != 0; }
+ public boolean hasArrivalTime() { return arrivalTime != null && !arrivalTime.isEmpty(); }
@JsonProperty("scanTime")
- public long getscanTime() { return scanTime;}
+ public String getscanTime() { return scanTime;}
- public void setScanTime(long scanTime) {
+ public void setScanTime(String scanTime) {
this.scanTime = scanTime;
}
- public boolean hasScanTime() { return scanTime != 0; }
+ public boolean hasScanTime() { return scanTime != null && !scanTime.isEmpty(); }
@JsonIgnore
public void setIsSystemMessage(boolean isSystemMessage) {
diff --git a/samza-sql/src/main/java/org/apache/samza/sql/translator/FilterTranslator.java b/samza-sql/src/main/java/org/apache/samza/sql/translator/FilterTranslator.java
index 18b73c8..ccab15a 100644
--- a/samza-sql/src/main/java/org/apache/samza/sql/translator/FilterTranslator.java
+++ b/samza-sql/src/main/java/org/apache/samza/sql/translator/FilterTranslator.java
@@ -19,6 +19,9 @@
package org.apache.samza.sql.translator;
+import com.google.common.annotations.VisibleForTesting;
+import java.time.Duration;
+import java.time.Instant;
import java.util.Arrays;
import java.util.Collections;
import org.apache.calcite.rel.logical.LogicalFilter;
@@ -32,7 +35,6 @@ import org.apache.samza.operators.functions.FilterFunction;
import org.apache.samza.sql.data.Expression;
import org.apache.samza.sql.data.SamzaSqlRelMessage;
import org.apache.samza.sql.runner.SamzaSqlApplicationContext;
-import org.apache.samza.sql.util.TranslatorUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -94,7 +96,7 @@ class FilterTranslator {
@Override
public boolean apply(SamzaSqlRelMessage message) {
- long startProcessing = System.nanoTime();
+ Instant startProcessing = Instant.now();
Object[] result = new Object[1];
expr.execute(translatorContext.getExecutionContext(), context, translatorContext.getDataContext(),
message.getSamzaSqlRelRecord().getFieldValues().toArray(), result);
@@ -103,7 +105,7 @@ class FilterTranslator {
log.debug(
String.format("return value for input %s is %s",
Arrays.asList(message.getSamzaSqlRelRecord().getFieldValues()).toString(), retVal));
- updateMetrics(startProcessing, retVal, System.nanoTime());
+ updateMetrics(startProcessing, retVal, Instant.now());
return retVal;
} else {
log.error("return value is not boolean");
@@ -113,17 +115,17 @@ class FilterTranslator {
/**
* Updates the MetricsRegistery of this operator
- * @param startProcessing = nanoTime when processing of the message started
- * @param endProcessing = nanoTIme when processing of the message ended
+ * @param startProcessing = begin processing of the message
+ * @param endProcessing = end of processing
*/
- private void updateMetrics(long startProcessing, boolean isOutput, long endProcessing) {
+ private void updateMetrics(Instant startProcessing, boolean isOutput, Instant endProcessing) {
inputEvents.inc();
if (isOutput) {
outputEvents.inc();
} else {
filteredOutEvents.inc();
}
- processingTime.update(TranslatorUtils.getMicroDurationFromNanoTimes(startProcessing, endProcessing));
+ processingTime.update(Duration.between(startProcessing, endProcessing).toMillis());
}
}
diff --git a/samza-sql/src/main/java/org/apache/samza/sql/translator/LogicalAggregateTranslator.java b/samza-sql/src/main/java/org/apache/samza/sql/translator/LogicalAggregateTranslator.java
index cc83857..aaa6a9e 100644
--- a/samza-sql/src/main/java/org/apache/samza/sql/translator/LogicalAggregateTranslator.java
+++ b/samza-sql/src/main/java/org/apache/samza/sql/translator/LogicalAggregateTranslator.java
@@ -20,6 +20,7 @@
package org.apache.samza.sql.translator;
import java.time.Duration;
+import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import org.apache.calcite.rel.logical.LogicalAggregate;
@@ -80,7 +81,7 @@ class LogicalAggregateTranslator {
List<Object> fieldValues = windowPane.getKey().getKey().getSamzaSqlRelRecord().getFieldValues();
fieldNames.add(aggFieldNames.get(0));
fieldValues.add(windowPane.getMessage());
- return new SamzaSqlRelMessage(fieldNames, fieldValues, new SamzaSqlRelMsgMetadata(0, 0,0));
+ return new SamzaSqlRelMessage(fieldNames, fieldValues, new SamzaSqlRelMsgMetadata("", "", ""));
});
context.registerMessageStream(aggregate.getId(), outputStream);
outputStream.map(new TranslatorOutputMetricsMapFunction(logicalOpId));
diff --git a/samza-sql/src/main/java/org/apache/samza/sql/translator/ProjectTranslator.java b/samza-sql/src/main/java/org/apache/samza/sql/translator/ProjectTranslator.java
index 0e49791..5b0dcc7 100644
--- a/samza-sql/src/main/java/org/apache/samza/sql/translator/ProjectTranslator.java
+++ b/samza-sql/src/main/java/org/apache/samza/sql/translator/ProjectTranslator.java
@@ -19,6 +19,8 @@
package org.apache.samza.sql.translator;
+import java.time.Duration;
+import java.time.Instant;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -42,7 +44,6 @@ import org.apache.samza.sql.data.Expression;
import org.apache.samza.sql.data.SamzaSqlRelMessage;
import org.apache.samza.sql.data.SamzaSqlRelMsgMetadata;
import org.apache.samza.sql.runner.SamzaSqlApplicationContext;
-import org.apache.samza.sql.util.TranslatorUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -112,7 +113,7 @@ class ProjectTranslator {
*/
@Override
public SamzaSqlRelMessage apply(SamzaSqlRelMessage message) {
- long arrivalTime = System.nanoTime();
+ Instant arrivalTime = Instant.now();
RelDataType type = project.getRowType();
Object[] output = new Object[type.getFieldCount()];
expr.execute(translatorContext.getExecutionContext(), context, translatorContext.getDataContext(),
@@ -121,22 +122,22 @@ class ProjectTranslator {
for (int index = 0; index < output.length; index++) {
names.add(index, project.getNamedProjects().get(index).getValue());
}
- updateMetrics(arrivalTime, System.nanoTime(), message.getSamzaSqlRelMsgMetadata().isNewInputMessage);
+ updateMetrics(arrivalTime, Instant.now(), message.getSamzaSqlRelMsgMetadata().isNewInputMessage);
return new SamzaSqlRelMessage(names, Arrays.asList(output), message.getSamzaSqlRelMsgMetadata());
}
/**
* Updates the Diagnostics Metrics (processing time and number of events)
- * @param arrivalTime input message arrival nanoTime (= begin of processing in this operator)
- * @param outputTime output message output nanoTime (=end of processing in this operator)
+ * @param arrivalTime input message arrival time (= beging of processing in this operator)
+ * @param outputTime output message output time (=end of processing in this operator)
* @param isNewInputMessage whether the input Message is from new input message or not
*/
- private void updateMetrics(long arrivalTime, long outputTime, boolean isNewInputMessage) {
+ private void updateMetrics(Instant arrivalTime, Instant outputTime, boolean isNewInputMessage) {
if (isNewInputMessage) {
inputEvents.inc();
}
outputEvents.inc();
- processingTime.update(TranslatorUtils.getMicroDurationFromNanoTimes(arrivalTime, outputTime));
+ processingTime.update(Duration.between(arrivalTime, outputTime).toMillis());
}
}
diff --git a/samza-sql/src/main/java/org/apache/samza/sql/translator/QueryTranslator.java b/samza-sql/src/main/java/org/apache/samza/sql/translator/QueryTranslator.java
index 4e818e4..2460b1f 100644
--- a/samza-sql/src/main/java/org/apache/samza/sql/translator/QueryTranslator.java
+++ b/samza-sql/src/main/java/org/apache/samza/sql/translator/QueryTranslator.java
@@ -20,6 +20,8 @@
package org.apache.samza.sql.translator;
import com.google.common.annotations.VisibleForTesting;
+import java.time.Duration;
+import java.time.Instant;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
@@ -61,7 +63,6 @@ import org.apache.samza.sql.planner.QueryPlanner;
import org.apache.samza.sql.runner.SamzaSqlApplicationConfig;
import org.apache.samza.sql.runner.SamzaSqlApplicationContext;
import org.apache.samza.sql.util.SamzaSqlQueryParser;
-import org.apache.samza.sql.util.TranslatorUtils;
import org.apache.samza.system.descriptors.DelegatingSystemDescriptor;
import org.apache.samza.system.descriptors.GenericOutputDescriptor;
import org.apache.samza.table.Table;
@@ -133,14 +134,14 @@ public class QueryTranslator {
queryLatency = new SamzaHistogram(metricsRegistry, queryLogicalId, TranslatorConstants.QUERY_LATENCY_NAME);
queueingLatency = new SamzaHistogram(metricsRegistry, queryLogicalId, TranslatorConstants.QUEUEING_LATENCY_NAME);
-
+
queryOutputEvents = metricsRegistry.newCounter(queryLogicalId, TranslatorConstants.OUTPUT_EVENTS_NAME);
queryOutputEvents.clear();
}
@Override
public KV<Object, Object> apply(SamzaSqlRelMessage message) {
- long beginProcessing = System.nanoTime();
+ Instant beginProcessing = Instant.now();
KV<Object, Object> retKV = this.samzaMsgConverter.convertToSamzaMessage(message);
if (message.getSamzaSqlRelRecord().containsField(SamzaSqlRelMessage.OP_NAME)
&& ((String) message.getSamzaSqlRelRecord().getField(SamzaSqlRelMessage.OP_NAME).get()).equalsIgnoreCase(
@@ -148,36 +149,36 @@ public class QueryTranslator {
// If it is a delete op. Set the payload to null so that the record gets deleted.
retKV = new KV<>(retKV.key, null);
}
- updateMetrics(beginProcessing, System.nanoTime(), message.getSamzaSqlRelMsgMetadata());
+ updateMetrics(beginProcessing, Instant.now(), message.getSamzaSqlRelMsgMetadata());
return retKV;
}
/**
* Updates the Diagnostics Metrics (processing time and number of events)
- * @param beginProcessing when (nanoTime) sendOutput Started processing this message
- * @param endProcessing when (nanoTime) sendOutput finished processing this message
+ * @param beginProcessing when sendOutput Started processing this message
+ * @param endProcessing when sendOutput finished processing this message
* @param metadata the event's message metadata
*/
- private void updateMetrics(long beginProcessing, long endProcessing, SamzaSqlRelMsgMetadata metadata) {
+ private void updateMetrics(Instant beginProcessing, Instant endProcessing, SamzaSqlRelMsgMetadata metadata) {
/* insert (SendToOutputStream) metrics */
- insertProcessingTime.update(TranslatorUtils.getMicroDurationFromNanoTimes(beginProcessing, endProcessing));
+ insertProcessingTime.update(Duration.between(beginProcessing, endProcessing).toMillis());
/* query metrics */
- long outputTime = System.nanoTime();
+ Instant outputTime = Instant.now();
queryOutputEvents.inc();
/* TODO: remove scanTime validation once code to assign it is stable */
Validate.isTrue(metadata.hasScanTime());
- long scanTime = metadata.getscanTime();
- queryLatency.update(TranslatorUtils.getMicroDurationFromNanoTimes(scanTime, outputTime));
+ Instant scanTime = Instant.parse(metadata.getscanTime());
+ queryLatency.update(Duration.between(scanTime, outputTime).toMillis());
/** TODO: change if hasArrivalTime to validation once arrivalTime is assigned,
and later remove the check once code is stable */
if (metadata.hasArrivalTime()) {
- long arrivalTime = metadata.getarrivalTime();
- queueingLatency.update(TranslatorUtils.getMicroDurationFromNanoTimes(arrivalTime, scanTime));
+ Instant arrivalTime = Instant.parse(metadata.getarrivalTime());
+ queueingLatency.update(Duration.between(arrivalTime, scanTime).toMillis());
}
/* since availability of eventTime depends on source, we need the following check */
if (metadata.hasEventTime()) {
- long eventTime = metadata.getEventTime();
- totalLatency.update(TranslatorUtils.getMicroDurationFromNanoTimes(eventTime, outputTime));
+ Instant eventTime = Instant.parse(metadata.getEventTime());
+ totalLatency.update(Duration.between(eventTime, outputTime).toMillis());
}
}
} // OutputMapFunction
diff --git a/samza-sql/src/main/java/org/apache/samza/sql/translator/ScanTranslator.java b/samza-sql/src/main/java/org/apache/samza/sql/translator/ScanTranslator.java
index ceee6d2..5fa04d8 100644
--- a/samza-sql/src/main/java/org/apache/samza/sql/translator/ScanTranslator.java
+++ b/samza-sql/src/main/java/org/apache/samza/sql/translator/ScanTranslator.java
@@ -19,6 +19,8 @@
package org.apache.samza.sql.translator;
+import java.time.Duration;
+import java.time.Instant;
import java.util.List;
import java.util.Map;
import org.apache.calcite.rel.core.TableScan;
@@ -40,7 +42,6 @@ import org.apache.samza.sql.data.SamzaSqlRelMessage;
import org.apache.samza.sql.interfaces.SamzaRelConverter;
import org.apache.samza.sql.interfaces.SqlIOConfig;
import org.apache.samza.sql.runner.SamzaSqlApplicationContext;
-import org.apache.samza.sql.util.TranslatorUtils;
import org.apache.samza.system.descriptors.DelegatingSystemDescriptor;
import org.apache.samza.system.descriptors.InputDescriptor;
import org.apache.samza.system.descriptors.InputTransformer;
@@ -127,25 +128,25 @@ class ScanTranslator {
@Override
public SamzaSqlRelMessage apply(SamzaSqlInputMessage samzaSqlInputMessage) {
- long startProcessing = System.nanoTime();
+ Instant startProcessing = Instant.now();
/* SAMZA-2089/LISAMZA-10654: the SamzaRelConverter.convertToRelMessage currently does not initialize
* the samzaSqlRelMessage.samzaSqlRelMsgMetadata, this needs to be fixed */
SamzaSqlRelMessage retMsg = this.msgConverter.convertToRelMessage(samzaSqlInputMessage.getKeyAndMessageKV());
retMsg.setEventTime(samzaSqlInputMessage.getMetadata().getEventTime());
retMsg.setArrivalTime(samzaSqlInputMessage.getMetadata().getarrivalTime());
- retMsg.setScanTime(startProcessing);
- updateMetrics(startProcessing, System.nanoTime());
+ retMsg.setScanTime(startProcessing.toString());
+ updateMetrics(startProcessing, Instant.now());
return retMsg;
}
/**
* Updates the MetricsRegistery of this operator
- * @param startProcessing = begin (nanoTime) processing of the message
- * @param endProcessing = end (nanoTime) of processing
+ * @param startProcessing = begin processing of the message
+ * @param endProcessing = end of processing
*/
- private void updateMetrics(long startProcessing, long endProcessing) {
+ private void updateMetrics(Instant startProcessing, Instant endProcessing) {
queryInputEvents.inc();
- processingTime.update(TranslatorUtils.getMicroDurationFromNanoTimes(startProcessing, endProcessing));
+ processingTime.update(Duration.between(startProcessing, endProcessing).toMillis());
}
} // ScanMapFunction
diff --git a/samza-sql/src/main/java/org/apache/samza/sql/translator/TranslatorConstants.java b/samza-sql/src/main/java/org/apache/samza/sql/translator/TranslatorConstants.java
index d788297..c2f9ea5 100644
--- a/samza-sql/src/main/java/org/apache/samza/sql/translator/TranslatorConstants.java
+++ b/samza-sql/src/main/java/org/apache/samza/sql/translator/TranslatorConstants.java
@@ -20,10 +20,10 @@
package org.apache.samza.sql.translator;
public class TranslatorConstants {
- public static final String PROCESSING_TIME_NAME = "processingTimeMicros";
- public static final String TOTAL_LATENCY_NAME = "totalLatencyMicros";
- public static final String QUERY_LATENCY_NAME = "queryLatencyMicros";
- public static final String QUEUEING_LATENCY_NAME = "queueingLatencyMicros";
+ public static final String PROCESSING_TIME_NAME = "processingTimeMs";
+ public static final String TOTAL_LATENCY_NAME = "totalLatencyMs";
+ public static final String QUERY_LATENCY_NAME = "queryLatencyMs";
+ public static final String QUEUEING_LATENCY_NAME = "queueingLatencyMs";
public static final String INPUT_EVENTS_NAME = "inputEvents";
public static final String FILTERED_EVENTS_NAME = "filteredEvents";
public static final String OUTPUT_EVENTS_NAME = "outputEvents";
diff --git a/samza-sql/src/main/java/org/apache/samza/sql/util/TranslatorUtils.java b/samza-sql/src/main/java/org/apache/samza/sql/util/TranslatorUtils.java
deleted file mode 100644
index c9d0239..0000000
--- a/samza-sql/src/main/java/org/apache/samza/sql/util/TranslatorUtils.java
+++ /dev/null
@@ -1,33 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.samza.sql.util;
-
-public class TranslatorUtils {
- static long MICROS_IN_NANO = 1000;
- /**
- * computes duraiton and converts it to micro resolution
- * @param start start nanoTime
- * @param end end nanoTime
- * @return duration in micro resolution
- */
- static public long getMicroDurationFromNanoTimes(long start, long end) {
- return (end - start) / MICROS_IN_NANO;
- }
-}
diff --git a/samza-sql/src/test/java/org/apache/samza/sql/data/TestSamzaSqlRelMessage.java b/samza-sql/src/test/java/org/apache/samza/sql/data/TestSamzaSqlRelMessage.java
index ac8e10d..dffb77e 100644
--- a/samza-sql/src/test/java/org/apache/samza/sql/data/TestSamzaSqlRelMessage.java
+++ b/samza-sql/src/test/java/org/apache/samza/sql/data/TestSamzaSqlRelMessage.java
@@ -36,34 +36,33 @@ public class TestSamzaSqlRelMessage {
@Test
public void testGetField() {
- SamzaSqlRelMessage message = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata(0, 0, 0));
+ SamzaSqlRelMessage message = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata("", "", ""));
Assert.assertEquals(values.get(0), message.getSamzaSqlRelRecord().getField(names.get(0)).get());
Assert.assertEquals(values.get(1), message.getSamzaSqlRelRecord().getField(names.get(1)).get());
}
@Test
public void testGetNonExistentField() {
- SamzaSqlRelMessage message = new SamzaSqlRelMessage(names, values,
- new SamzaSqlRelMsgMetadata(0, 0, 0));
+ SamzaSqlRelMessage message = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata("", "", ""));
Assert.assertFalse(message.getSamzaSqlRelRecord().getField("field3").isPresent());
}
@Test
public void testEquality() {
- SamzaSqlRelMessage message1 = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata(0, 0, 0));
+ SamzaSqlRelMessage message1 = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata("", "", ""));
SamzaSqlRelMessage message2 =
new SamzaSqlRelMessage(Arrays.asList("field1", "field2"), Arrays.asList("value1", "value2"),
- new SamzaSqlRelMsgMetadata(0, 0, 0));
+ new SamzaSqlRelMsgMetadata("", "", ""));
Assert.assertEquals(message1, message2);
Assert.assertEquals(message1.hashCode(), message2.hashCode());
}
@Test
public void testInEquality() {
- SamzaSqlRelMessage message1 = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata(0, 0, 0));
+ SamzaSqlRelMessage message1 = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata("", "", ""));
SamzaSqlRelMessage message2 =
new SamzaSqlRelMessage(Arrays.asList("field1", "field2"), Arrays.asList("value2", "value2"),
- new SamzaSqlRelMsgMetadata(0, 0, 0));
+ new SamzaSqlRelMsgMetadata("", "", ""));
Assert.assertNotEquals(message1, message2);
Assert.assertNotEquals(message1.hashCode(), message2.hashCode());
}
@@ -71,7 +70,7 @@ public class TestSamzaSqlRelMessage {
@Test
public void testCompositeKeyCreation() {
List<String> keyPartNames = Arrays.asList("kfield1", "kfield2");
- SamzaSqlRelMessage message = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata(0, 0, 0));
+ SamzaSqlRelMessage message = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata("", "", ""));
SamzaSqlRelRecord relRecord1 = SamzaSqlRelMessage.createSamzaSqlCompositeKey(message, Collections.singletonList(0));
Assert.assertEquals(relRecord1.getFieldNames().size(), 1);
@@ -90,7 +89,7 @@ public class TestSamzaSqlRelMessage {
@Test (expected = IllegalArgumentException.class)
public void testCompositeKeyCreationWithInEqualKeyNameValues() {
List<String> keyPartNames = Arrays.asList("kfield1", "kfield2");
- SamzaSqlRelMessage message = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata(0, 0, 0));
+ SamzaSqlRelMessage message = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata("", "", ""));
SamzaSqlRelRecord relRecord1 = SamzaSqlRelMessage.createSamzaSqlCompositeKey(message, Arrays.asList(1, 0),
SamzaSqlRelMessage.getSamzaSqlCompositeKeyFieldNames(keyPartNames, Arrays.asList(1)));
diff --git a/samza-sql/src/test/java/org/apache/samza/sql/serializers/TestSamzaSqlRelMessageSerde.java b/samza-sql/src/test/java/org/apache/samza/sql/serializers/TestSamzaSqlRelMessageSerde.java
index 689e8d2..5afa4fe 100644
--- a/samza-sql/src/test/java/org/apache/samza/sql/serializers/TestSamzaSqlRelMessageSerde.java
+++ b/samza-sql/src/test/java/org/apache/samza/sql/serializers/TestSamzaSqlRelMessageSerde.java
@@ -53,8 +53,7 @@ public class TestSamzaSqlRelMessageSerde {
@Test
public void testWithDifferentFields() {
- SamzaSqlRelMessage message = new SamzaSqlRelMessage(names, values,
- new SamzaSqlRelMsgMetadata(0, 0, 0));
+ SamzaSqlRelMessage message = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata("", "", ""));
SamzaSqlRelMessageSerde serde =
(SamzaSqlRelMessageSerde) new SamzaSqlRelMessageSerdeFactory().getSerde(null, null);
SamzaSqlRelMessage resultMsg = serde.fromBytes(serde.toBytes(message));
diff --git a/samza-sql/src/test/java/org/apache/samza/sql/serializers/TestSamzaSqlRelRecordSerde.java b/samza-sql/src/test/java/org/apache/samza/sql/serializers/TestSamzaSqlRelRecordSerde.java
index 6885fc1..4e6ff4f 100644
--- a/samza-sql/src/test/java/org/apache/samza/sql/serializers/TestSamzaSqlRelRecordSerde.java
+++ b/samza-sql/src/test/java/org/apache/samza/sql/serializers/TestSamzaSqlRelRecordSerde.java
@@ -51,7 +51,7 @@ public class TestSamzaSqlRelRecordSerde {
@Test
public void testWithDifferentFields() {
SamzaSqlRelRecord record =
- new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata(0,0, 0)).getSamzaSqlRelRecord();
+ new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata("", "", "")).getSamzaSqlRelRecord();
SamzaSqlRelRecordSerde serde =
(SamzaSqlRelRecordSerde) new SamzaSqlRelRecordSerdeFactory().getSerde(null, null);
SamzaSqlRelRecord resultRecord = serde.fromBytes(serde.toBytes(record));
diff --git a/samza-sql/src/test/java/org/apache/samza/sql/translator/TestFilterTranslator.java b/samza-sql/src/test/java/org/apache/samza/sql/translator/TestFilterTranslator.java
index be64cc4..2a98c8c 100644
--- a/samza-sql/src/test/java/org/apache/samza/sql/translator/TestFilterTranslator.java
+++ b/samza-sql/src/test/java/org/apache/samza/sql/translator/TestFilterTranslator.java
@@ -125,7 +125,7 @@ public class TestFilterTranslator extends TranslatorTestBase {
// Calling filterFn.apply() to verify the filter function is correctly applied to the input message
SamzaSqlRelMessage mockInputMsg = new SamzaSqlRelMessage(new ArrayList<>(), new ArrayList<>(),
- new SamzaSqlRelMsgMetadata(0, 0, 0));
+ new SamzaSqlRelMsgMetadata("", "", ""));
SamzaSqlExecutionContext executionContext = mock(SamzaSqlExecutionContext.class);
DataContext dataContext = mock(DataContext.class);
when(mockTranslatorContext.getExecutionContext()).thenReturn(executionContext);
diff --git a/samza-sql/src/test/java/org/apache/samza/sql/translator/TestProjectTranslator.java b/samza-sql/src/test/java/org/apache/samza/sql/translator/TestProjectTranslator.java
index 829532c..3a2b943 100644
--- a/samza-sql/src/test/java/org/apache/samza/sql/translator/TestProjectTranslator.java
+++ b/samza-sql/src/test/java/org/apache/samza/sql/translator/TestProjectTranslator.java
@@ -143,7 +143,7 @@ public class TestProjectTranslator extends TranslatorTestBase {
// Calling mapFn.apply() to verify the filter function is correctly applied to the input message
SamzaSqlRelMessage mockInputMsg = new SamzaSqlRelMessage(new ArrayList<>(), new ArrayList<>(),
- new SamzaSqlRelMsgMetadata(0, 0, 0));
+ new SamzaSqlRelMsgMetadata("", "", ""));
SamzaSqlExecutionContext executionContext = mock(SamzaSqlExecutionContext.class);
DataContext dataContext = mock(DataContext.class);
when(mockTranslatorContext.getExecutionContext()).thenReturn(executionContext);
diff --git a/samza-sql/src/test/java/org/apache/samza/sql/translator/TestSamzaSqlLocalTableJoinFunction.java b/samza-sql/src/test/java/org/apache/samza/sql/translator/TestSamzaSqlLocalTableJoinFunction.java
index b92dfe8..53c7bc1 100644
--- a/samza-sql/src/test/java/org/apache/samza/sql/translator/TestSamzaSqlLocalTableJoinFunction.java
+++ b/samza-sql/src/test/java/org/apache/samza/sql/translator/TestSamzaSqlLocalTableJoinFunction.java
@@ -44,10 +44,8 @@ public class TestSamzaSqlLocalTableJoinFunction {
@Test
public void testWithInnerJoinWithTableOnRight() {
- SamzaSqlRelMessage streamMsg = new SamzaSqlRelMessage(streamFieldNames, streamFieldValues,
- new SamzaSqlRelMsgMetadata(0, 0, 0));
- SamzaSqlRelMessage tableMsg = new SamzaSqlRelMessage(tableFieldNames, tableFieldValues,
- new SamzaSqlRelMsgMetadata(0, 0, 0));
+ SamzaSqlRelMessage streamMsg = new SamzaSqlRelMessage(streamFieldNames, streamFieldValues, new SamzaSqlRelMsgMetadata("", "", ""));
+ SamzaSqlRelMessage tableMsg = new SamzaSqlRelMessage(tableFieldNames, tableFieldValues, new SamzaSqlRelMsgMetadata("", "", ""));
JoinRelType joinRelType = JoinRelType.INNER;
List<Integer> streamKeyIds = Arrays.asList(0, 1);
List<Integer> tableKeyIds = Arrays.asList(0, 1);
@@ -79,10 +77,8 @@ public class TestSamzaSqlLocalTableJoinFunction {
@Test
public void testWithInnerJoinWithTableOnLeft() {
- SamzaSqlRelMessage streamMsg = new SamzaSqlRelMessage(streamFieldNames, streamFieldValues,
- new SamzaSqlRelMsgMetadata(0, 0, 0));
- SamzaSqlRelMessage tableMsg = new SamzaSqlRelMessage(tableFieldNames, tableFieldValues,
- new SamzaSqlRelMsgMetadata(0, 0, 0));
+ SamzaSqlRelMessage streamMsg = new SamzaSqlRelMessage(streamFieldNames, streamFieldValues, new SamzaSqlRelMsgMetadata("", "", ""));
+ SamzaSqlRelMessage tableMsg = new SamzaSqlRelMessage(tableFieldNames, tableFieldValues, new SamzaSqlRelMsgMetadata("", "", ""));
JoinRelType joinRelType = JoinRelType.INNER;
List<Integer> streamKeyIds = Arrays.asList(0, 2);
List<Integer> tableKeyIds = Arrays.asList(0, 2);
@@ -115,7 +111,7 @@ public class TestSamzaSqlLocalTableJoinFunction {
@Test
public void testNullRecordWithInnerJoin() {
SamzaSqlRelMessage streamMsg = new SamzaSqlRelMessage(streamFieldNames, streamFieldValues,
- new SamzaSqlRelMsgMetadata(0, 0, 0));
+ new SamzaSqlRelMsgMetadata("", "", ""));
JoinRelType joinRelType = JoinRelType.INNER;
List<Integer> streamKeyIds = Arrays.asList(0, 1);
List<Integer> tableKeyIds = Arrays.asList(2, 3);
@@ -139,7 +135,7 @@ public class TestSamzaSqlLocalTableJoinFunction {
@Test
public void testNullRecordWithLeftOuterJoin() {
SamzaSqlRelMessage streamMsg = new SamzaSqlRelMessage(streamFieldNames, streamFieldValues,
- new SamzaSqlRelMsgMetadata(0, 0, 0));
+ new SamzaSqlRelMsgMetadata("", "", ""));
JoinRelType joinRelType = JoinRelType.LEFT;
List<Integer> streamKeyIds = Arrays.asList(0, 1);
List<Integer> tableKeyIds = Arrays.asList(2, 3);
diff --git a/samza-sql/src/test/java/org/apache/samza/sql/translator/TestSamzaSqlRemoteTableJoinFunction.java b/samza-sql/src/test/java/org/apache/samza/sql/translator/TestSamzaSqlRemoteTableJoinFunction.java
index 20cea14..2deb6ad 100644
--- a/samza-sql/src/test/java/org/apache/samza/sql/translator/TestSamzaSqlRemoteTableJoinFunction.java
+++ b/samza-sql/src/test/java/org/apache/samza/sql/translator/TestSamzaSqlRemoteTableJoinFunction.java
@@ -69,7 +69,7 @@ public class TestSamzaSqlRemoteTableJoinFunction {
tableRecord.put("name", "name1");
SamzaSqlRelMessage streamMsg = new SamzaSqlRelMessage(streamFieldNames, streamFieldValues,
- new SamzaSqlRelMsgMetadata(0, 0, 0));
+ new SamzaSqlRelMsgMetadata("", "", ""));
SamzaSqlRelMessage tableMsg = relConverter.convertToRelMessage(new KV(tableRecord.get("id"), tableRecord));
JoinRelType joinRelType = JoinRelType.INNER;
List<Integer> streamKeyIds = Arrays.asList(1);