You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by at...@apache.org on 2019/07/26 15:42:31 UTC
[samza] branch master updated: Fixing SamzaSQL Latency Metrics
resolution (#1116)
This is an automated email from the ASF dual-hosted git repository.
atoomula 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 ec536e1 Fixing SamzaSQL Latency Metrics resolution (#1116)
ec536e1 is described below
commit ec536e14a1aa3d0fbd31c76418b605ea82c8b8d9
Author: Shenoda Guirguis <sg...@linkedin.com>
AuthorDate: Fri Jul 26 08:42:26 2019 -0700
Fixing SamzaSQL Latency Metrics resolution (#1116)
---
.../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 +++---
.../TranslatorUtils.java} | 23 ++++++++--------
.../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, 104 insertions(+), 105 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 7392253..e8bb331 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 timestamp when this event occured, should be set by the event source, 0 means unassigned
+ // the System's nanoTime when this event occurred, should be set by the event source, 0 means unassigned
private long eventTime = 0L;
- // the timestamp when this event is pickedup by samza, 0 means unassgined
- private long arrivalTime = 0L;
+ // the System's nanoTime when this event is picked-up by samza, 0 means unassgined
+ private long arrivalTime;
/**
* 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 = Instant.now().toEpochMilli();
+ this.arrivalTime = System.nanoTime();
}
/**
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 2bbce22..6bcae8b 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,9 +40,7 @@ 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() == 0) ? "" : Instant.ofEpochMilli(ime.getEventTime()).toString(),
- (ime.getArrivalTime() == 0) ? "" : Instant.ofEpochMilli(ime.getArrivalTime()).toString(), null);
+ SamzaSqlRelMsgMetadata metadata = new SamzaSqlRelMsgMetadata(ime.getEventTime(), ime.getArrivalTime(),0L);
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 fcb0c72..3d3ebe2 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("", "", ""));
+ new SamzaSqlRelMsgMetadata(0, 0, 0));
}
/**
@@ -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("", "", ""));
+ new SamzaSqlRelMsgMetadata(0, 0, 0));
}
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 d538a8e..84db61f 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(String eventTime) {
+ public void setEventTime(long eventTime) {
this.samzaSqlRelMsgMetadata.setEventTime(eventTime);
}
- public void setArrivalTime(String arrivalTime) {
+ public void setArrivalTime(long arrivalTime) {
this.samzaSqlRelMsgMetadata.setArrivalTime(arrivalTime);
}
- public void setScanTime(String scanTime) {
+ public void setScanTime(long 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 14f2892..60a4b11 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 String eventTime;
+ private long eventTime;
/**
* the timestamp of when Samza App received the event
* TODO: set arrivalTime during conversion from IME to SamzaMessage
*/
@JsonProperty("arrivalTime")
- private String arrivalTime;
+ private long arrivalTime;
/**
* the timestamp when SamzaSQL query starts processing the event
* set by the SamzaSQL Scan operator
*/
@JsonProperty("scanTime")
- private String scanTime;
+ private long scanTime;
- public SamzaSqlRelMsgMetadata(@JsonProperty("eventTime") String eventTime, @JsonProperty("arrivalTime") String arrivalTime,
- @JsonProperty("scanTime") String scanTime) {
+ public SamzaSqlRelMsgMetadata(@JsonProperty("eventTime") long eventTime, @JsonProperty("arrivalTime") long arrivalTime,
+ @JsonProperty("scanTime") long scanTime) {
this.eventTime = eventTime;
this.arrivalTime = arrivalTime;
this.scanTime = scanTime;
}
- public SamzaSqlRelMsgMetadata(String eventTime, String arrivalTime, String scanTime, boolean isNewInputMessage) {
+ public SamzaSqlRelMsgMetadata(long eventTime, long arrivalTime, long scanTime, boolean isNewInputMessage) {
this(eventTime, arrivalTime, scanTime);
this.isNewInputMessage = isNewInputMessage;
}
@JsonProperty("eventTime")
- public String getEventTime() { return eventTime;}
+ public long getEventTime() { return eventTime;}
- public void setEventTime(String eventTime) {
+ public void setEventTime(long eventTime) {
this.eventTime = eventTime;
}
- public boolean hasEventTime() { return eventTime != null && !eventTime.isEmpty(); }
+ public boolean hasEventTime() { return eventTime != 0; }
@JsonProperty("arrivalTime")
- public String getarrivalTime() { return arrivalTime;}
+ public long getarrivalTime() { return arrivalTime;}
- public void setArrivalTime(String arrivalTime) {
+ public void setArrivalTime(long arrivalTime) {
this.arrivalTime = arrivalTime;
}
- public boolean hasArrivalTime() { return arrivalTime != null && !arrivalTime.isEmpty(); }
+ public boolean hasArrivalTime() { return arrivalTime != 0; }
@JsonProperty("scanTime")
- public String getscanTime() { return scanTime;}
+ public long getscanTime() { return scanTime;}
- public void setScanTime(String scanTime) {
+ public void setScanTime(long scanTime) {
this.scanTime = scanTime;
}
- public boolean hasScanTime() { return scanTime != null && !scanTime.isEmpty(); }
+ public boolean hasScanTime() { return scanTime != 0; }
@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 ccab15a..18b73c8 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,9 +19,6 @@
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;
@@ -35,6 +32,7 @@ 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;
@@ -96,7 +94,7 @@ class FilterTranslator {
@Override
public boolean apply(SamzaSqlRelMessage message) {
- Instant startProcessing = Instant.now();
+ long startProcessing = System.nanoTime();
Object[] result = new Object[1];
expr.execute(translatorContext.getExecutionContext(), context, translatorContext.getDataContext(),
message.getSamzaSqlRelRecord().getFieldValues().toArray(), result);
@@ -105,7 +103,7 @@ class FilterTranslator {
log.debug(
String.format("return value for input %s is %s",
Arrays.asList(message.getSamzaSqlRelRecord().getFieldValues()).toString(), retVal));
- updateMetrics(startProcessing, retVal, Instant.now());
+ updateMetrics(startProcessing, retVal, System.nanoTime());
return retVal;
} else {
log.error("return value is not boolean");
@@ -115,17 +113,17 @@ class FilterTranslator {
/**
* Updates the MetricsRegistery of this operator
- * @param startProcessing = begin processing of the message
- * @param endProcessing = end of processing
+ * @param startProcessing = nanoTime when processing of the message started
+ * @param endProcessing = nanoTIme when processing of the message ended
*/
- private void updateMetrics(Instant startProcessing, boolean isOutput, Instant endProcessing) {
+ private void updateMetrics(long startProcessing, boolean isOutput, long endProcessing) {
inputEvents.inc();
if (isOutput) {
outputEvents.inc();
} else {
filteredOutEvents.inc();
}
- processingTime.update(Duration.between(startProcessing, endProcessing).toMillis());
+ processingTime.update(TranslatorUtils.getMicroDurationFromNanoTimes(startProcessing, endProcessing));
}
}
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 aaa6a9e..cc83857 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,7 +20,6 @@
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;
@@ -81,7 +80,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("", "", ""));
+ return new SamzaSqlRelMessage(fieldNames, fieldValues, new SamzaSqlRelMsgMetadata(0, 0,0));
});
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 5b0dcc7..0e49791 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,8 +19,6 @@
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;
@@ -44,6 +42,7 @@ 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;
@@ -113,7 +112,7 @@ class ProjectTranslator {
*/
@Override
public SamzaSqlRelMessage apply(SamzaSqlRelMessage message) {
- Instant arrivalTime = Instant.now();
+ long arrivalTime = System.nanoTime();
RelDataType type = project.getRowType();
Object[] output = new Object[type.getFieldCount()];
expr.execute(translatorContext.getExecutionContext(), context, translatorContext.getDataContext(),
@@ -122,22 +121,22 @@ class ProjectTranslator {
for (int index = 0; index < output.length; index++) {
names.add(index, project.getNamedProjects().get(index).getValue());
}
- updateMetrics(arrivalTime, Instant.now(), message.getSamzaSqlRelMsgMetadata().isNewInputMessage);
+ updateMetrics(arrivalTime, System.nanoTime(), 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 time (= beging of processing in this operator)
- * @param outputTime output message output time (=end of processing in this operator)
+ * @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 isNewInputMessage whether the input Message is from new input message or not
*/
- private void updateMetrics(Instant arrivalTime, Instant outputTime, boolean isNewInputMessage) {
+ private void updateMetrics(long arrivalTime, long outputTime, boolean isNewInputMessage) {
if (isNewInputMessage) {
inputEvents.inc();
}
outputEvents.inc();
- processingTime.update(Duration.between(arrivalTime, outputTime).toMillis());
+ processingTime.update(TranslatorUtils.getMicroDurationFromNanoTimes(arrivalTime, outputTime));
}
}
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 2460b1f..4e818e4 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,8 +20,6 @@
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;
@@ -63,6 +61,7 @@ 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;
@@ -134,14 +133,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) {
- Instant beginProcessing = Instant.now();
+ long beginProcessing = System.nanoTime();
KV<Object, Object> retKV = this.samzaMsgConverter.convertToSamzaMessage(message);
if (message.getSamzaSqlRelRecord().containsField(SamzaSqlRelMessage.OP_NAME)
&& ((String) message.getSamzaSqlRelRecord().getField(SamzaSqlRelMessage.OP_NAME).get()).equalsIgnoreCase(
@@ -149,36 +148,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, Instant.now(), message.getSamzaSqlRelMsgMetadata());
+ updateMetrics(beginProcessing, System.nanoTime(), message.getSamzaSqlRelMsgMetadata());
return retKV;
}
/**
* Updates the Diagnostics Metrics (processing time and number of events)
- * @param beginProcessing when sendOutput Started processing this message
- * @param endProcessing when sendOutput finished processing this message
+ * @param beginProcessing when (nanoTime) sendOutput Started processing this message
+ * @param endProcessing when (nanoTime) sendOutput finished processing this message
* @param metadata the event's message metadata
*/
- private void updateMetrics(Instant beginProcessing, Instant endProcessing, SamzaSqlRelMsgMetadata metadata) {
+ private void updateMetrics(long beginProcessing, long endProcessing, SamzaSqlRelMsgMetadata metadata) {
/* insert (SendToOutputStream) metrics */
- insertProcessingTime.update(Duration.between(beginProcessing, endProcessing).toMillis());
+ insertProcessingTime.update(TranslatorUtils.getMicroDurationFromNanoTimes(beginProcessing, endProcessing));
/* query metrics */
- Instant outputTime = Instant.now();
+ long outputTime = System.nanoTime();
queryOutputEvents.inc();
/* TODO: remove scanTime validation once code to assign it is stable */
Validate.isTrue(metadata.hasScanTime());
- Instant scanTime = Instant.parse(metadata.getscanTime());
- queryLatency.update(Duration.between(scanTime, outputTime).toMillis());
+ long scanTime = metadata.getscanTime();
+ queryLatency.update(TranslatorUtils.getMicroDurationFromNanoTimes(scanTime, outputTime));
/** TODO: change if hasArrivalTime to validation once arrivalTime is assigned,
and later remove the check once code is stable */
if (metadata.hasArrivalTime()) {
- Instant arrivalTime = Instant.parse(metadata.getarrivalTime());
- queueingLatency.update(Duration.between(arrivalTime, scanTime).toMillis());
+ long arrivalTime = metadata.getarrivalTime();
+ queueingLatency.update(TranslatorUtils.getMicroDurationFromNanoTimes(arrivalTime, scanTime));
}
/* since availability of eventTime depends on source, we need the following check */
if (metadata.hasEventTime()) {
- Instant eventTime = Instant.parse(metadata.getEventTime());
- totalLatency.update(Duration.between(eventTime, outputTime).toMillis());
+ long eventTime = metadata.getEventTime();
+ totalLatency.update(TranslatorUtils.getMicroDurationFromNanoTimes(eventTime, outputTime));
}
}
} // 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 5fa04d8..ceee6d2 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,8 +19,6 @@
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;
@@ -42,6 +40,7 @@ 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;
@@ -128,25 +127,25 @@ class ScanTranslator {
@Override
public SamzaSqlRelMessage apply(SamzaSqlInputMessage samzaSqlInputMessage) {
- Instant startProcessing = Instant.now();
+ long startProcessing = System.nanoTime();
/* 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.toString());
- updateMetrics(startProcessing, Instant.now());
+ retMsg.setScanTime(startProcessing);
+ updateMetrics(startProcessing, System.nanoTime());
return retMsg;
}
/**
* Updates the MetricsRegistery of this operator
- * @param startProcessing = begin processing of the message
- * @param endProcessing = end of processing
+ * @param startProcessing = begin (nanoTime) processing of the message
+ * @param endProcessing = end (nanoTime) of processing
*/
- private void updateMetrics(Instant startProcessing, Instant endProcessing) {
+ private void updateMetrics(long startProcessing, long endProcessing) {
queryInputEvents.inc();
- processingTime.update(Duration.between(startProcessing, endProcessing).toMillis());
+ processingTime.update(TranslatorUtils.getMicroDurationFromNanoTimes(startProcessing, endProcessing));
}
} // 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 c2f9ea5..d788297 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 = "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 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 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/translator/TranslatorConstants.java b/samza-sql/src/main/java/org/apache/samza/sql/util/TranslatorUtils.java
similarity index 54%
copy from samza-sql/src/main/java/org/apache/samza/sql/translator/TranslatorConstants.java
copy to samza-sql/src/main/java/org/apache/samza/sql/util/TranslatorUtils.java
index c2f9ea5..c9d0239 100644
--- a/samza-sql/src/main/java/org/apache/samza/sql/translator/TranslatorConstants.java
+++ b/samza-sql/src/main/java/org/apache/samza/sql/util/TranslatorUtils.java
@@ -17,16 +17,17 @@
* under the License.
*/
-package org.apache.samza.sql.translator;
+package org.apache.samza.sql.util;
-public class TranslatorConstants {
- 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";
- public static final String LOGOPID_TEMPLATE = "sql_%d_%s_%d";
- public static final String LOGSQLID_TEMPLATE = "sql_%d";
+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 dffb77e..ac8e10d 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,33 +36,34 @@ public class TestSamzaSqlRelMessage {
@Test
public void testGetField() {
- SamzaSqlRelMessage message = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata("", "", ""));
+ SamzaSqlRelMessage message = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata(0, 0, 0));
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("", "", ""));
+ SamzaSqlRelMessage message = new SamzaSqlRelMessage(names, values,
+ new SamzaSqlRelMsgMetadata(0, 0, 0));
Assert.assertFalse(message.getSamzaSqlRelRecord().getField("field3").isPresent());
}
@Test
public void testEquality() {
- SamzaSqlRelMessage message1 = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata("", "", ""));
+ SamzaSqlRelMessage message1 = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata(0, 0, 0));
SamzaSqlRelMessage message2 =
new SamzaSqlRelMessage(Arrays.asList("field1", "field2"), Arrays.asList("value1", "value2"),
- new SamzaSqlRelMsgMetadata("", "", ""));
+ new SamzaSqlRelMsgMetadata(0, 0, 0));
Assert.assertEquals(message1, message2);
Assert.assertEquals(message1.hashCode(), message2.hashCode());
}
@Test
public void testInEquality() {
- SamzaSqlRelMessage message1 = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata("", "", ""));
+ SamzaSqlRelMessage message1 = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata(0, 0, 0));
SamzaSqlRelMessage message2 =
new SamzaSqlRelMessage(Arrays.asList("field1", "field2"), Arrays.asList("value2", "value2"),
- new SamzaSqlRelMsgMetadata("", "", ""));
+ new SamzaSqlRelMsgMetadata(0, 0, 0));
Assert.assertNotEquals(message1, message2);
Assert.assertNotEquals(message1.hashCode(), message2.hashCode());
}
@@ -70,7 +71,7 @@ public class TestSamzaSqlRelMessage {
@Test
public void testCompositeKeyCreation() {
List<String> keyPartNames = Arrays.asList("kfield1", "kfield2");
- SamzaSqlRelMessage message = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata("", "", ""));
+ SamzaSqlRelMessage message = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata(0, 0, 0));
SamzaSqlRelRecord relRecord1 = SamzaSqlRelMessage.createSamzaSqlCompositeKey(message, Collections.singletonList(0));
Assert.assertEquals(relRecord1.getFieldNames().size(), 1);
@@ -89,7 +90,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("", "", ""));
+ SamzaSqlRelMessage message = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata(0, 0, 0));
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 5afa4fe..689e8d2 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,7 +53,8 @@ public class TestSamzaSqlRelMessageSerde {
@Test
public void testWithDifferentFields() {
- SamzaSqlRelMessage message = new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata("", "", ""));
+ SamzaSqlRelMessage message = new SamzaSqlRelMessage(names, values,
+ new SamzaSqlRelMsgMetadata(0, 0, 0));
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 4e6ff4f..6885fc1 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("", "", "")).getSamzaSqlRelRecord();
+ new SamzaSqlRelMessage(names, values, new SamzaSqlRelMsgMetadata(0,0, 0)).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 2a98c8c..be64cc4 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("", "", ""));
+ new SamzaSqlRelMsgMetadata(0, 0, 0));
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 3a2b943..829532c 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("", "", ""));
+ new SamzaSqlRelMsgMetadata(0, 0, 0));
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 53c7bc1..b92dfe8 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,8 +44,10 @@ public class TestSamzaSqlLocalTableJoinFunction {
@Test
public void testWithInnerJoinWithTableOnRight() {
- SamzaSqlRelMessage streamMsg = new SamzaSqlRelMessage(streamFieldNames, streamFieldValues, new SamzaSqlRelMsgMetadata("", "", ""));
- SamzaSqlRelMessage tableMsg = new SamzaSqlRelMessage(tableFieldNames, tableFieldValues, new SamzaSqlRelMsgMetadata("", "", ""));
+ SamzaSqlRelMessage streamMsg = new SamzaSqlRelMessage(streamFieldNames, streamFieldValues,
+ new SamzaSqlRelMsgMetadata(0, 0, 0));
+ SamzaSqlRelMessage tableMsg = new SamzaSqlRelMessage(tableFieldNames, tableFieldValues,
+ new SamzaSqlRelMsgMetadata(0, 0, 0));
JoinRelType joinRelType = JoinRelType.INNER;
List<Integer> streamKeyIds = Arrays.asList(0, 1);
List<Integer> tableKeyIds = Arrays.asList(0, 1);
@@ -77,8 +79,10 @@ public class TestSamzaSqlLocalTableJoinFunction {
@Test
public void testWithInnerJoinWithTableOnLeft() {
- SamzaSqlRelMessage streamMsg = new SamzaSqlRelMessage(streamFieldNames, streamFieldValues, new SamzaSqlRelMsgMetadata("", "", ""));
- SamzaSqlRelMessage tableMsg = new SamzaSqlRelMessage(tableFieldNames, tableFieldValues, new SamzaSqlRelMsgMetadata("", "", ""));
+ SamzaSqlRelMessage streamMsg = new SamzaSqlRelMessage(streamFieldNames, streamFieldValues,
+ new SamzaSqlRelMsgMetadata(0, 0, 0));
+ SamzaSqlRelMessage tableMsg = new SamzaSqlRelMessage(tableFieldNames, tableFieldValues,
+ new SamzaSqlRelMsgMetadata(0, 0, 0));
JoinRelType joinRelType = JoinRelType.INNER;
List<Integer> streamKeyIds = Arrays.asList(0, 2);
List<Integer> tableKeyIds = Arrays.asList(0, 2);
@@ -111,7 +115,7 @@ public class TestSamzaSqlLocalTableJoinFunction {
@Test
public void testNullRecordWithInnerJoin() {
SamzaSqlRelMessage streamMsg = new SamzaSqlRelMessage(streamFieldNames, streamFieldValues,
- new SamzaSqlRelMsgMetadata("", "", ""));
+ new SamzaSqlRelMsgMetadata(0, 0, 0));
JoinRelType joinRelType = JoinRelType.INNER;
List<Integer> streamKeyIds = Arrays.asList(0, 1);
List<Integer> tableKeyIds = Arrays.asList(2, 3);
@@ -135,7 +139,7 @@ public class TestSamzaSqlLocalTableJoinFunction {
@Test
public void testNullRecordWithLeftOuterJoin() {
SamzaSqlRelMessage streamMsg = new SamzaSqlRelMessage(streamFieldNames, streamFieldValues,
- new SamzaSqlRelMsgMetadata("", "", ""));
+ new SamzaSqlRelMsgMetadata(0, 0, 0));
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 2deb6ad..20cea14 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("", "", ""));
+ new SamzaSqlRelMsgMetadata(0, 0, 0));
SamzaSqlRelMessage tableMsg = relConverter.convertToRelMessage(new KV(tableRecord.get("id"), tableRecord));
JoinRelType joinRelType = JoinRelType.INNER;
List<Integer> streamKeyIds = Arrays.asList(1);