Propogating the system messages to the stream. (apache#937)

* Adding system messages to the stream

* checkstyle fixes

Author: srinipunuru

samza-sql/src/main/java/org/apache/samza/sql/data/SamzaSqlRelMsgMetadata.java

@@ -20,6 +20,8 @@
 package org.apache.samza.sql.data;
 
 import java.io.Serializable;
+import java.time.Instant;
+import org.codehaus.jackson.annotate.JsonIgnore;
 import org.codehaus.jackson.annotate.JsonProperty;
 
 
@@ -36,9 +38,16 @@ public class SamzaSqlRelMsgMetadata implements Serializable {
   public boolean isNewInputMessage = true;
 
   /**
-   *
+   * Indicates whether the SamzaSqlMessage is a system message or not.
    */
-  public String operatorBeginProcessingInstant = null;
+  @JsonIgnore
+  private boolean isSystemMessage = false;
+
+  /**
+   * Time at which the join operation started for the message.
+   * If there is no join node in the operator graph, this will be -1.
+   */
+  public long joinStartTimeMs = -1;
 
 
   /**
@@ -93,7 +102,6 @@ public void setArrivalTime(String arrivalTime) {
 
   public boolean hasArrivalTime() { return arrivalTime != null && !arrivalTime.isEmpty(); }
 
-
   @JsonProperty("scanTime")
   public String getscanTime() { return scanTime;}
 
@@ -103,6 +111,16 @@ public void setScanTime(String scanTime) {
 
   public boolean hasScanTime() { return scanTime != null && !scanTime.isEmpty(); }
 
+  @JsonIgnore
+  public  void setIsSystemMessage(boolean isSystemMessage) {
+    this.isSystemMessage = isSystemMessage;
+  }
+
+  @JsonIgnore
+  public boolean isSystemMessage() {
+    return isSystemMessage;
+  }
+
   @Override
   public String toString() {
     return "[Metadata:{" + eventTime + " " + arrivalTime + " " + scanTime + "}]";

samza-sql/src/main/java/org/apache/samza/sql/runner/SamzaSqlApplicationConfig.java

@@ -89,6 +89,7 @@ public class SamzaSqlApplicationConfig {
 
   public static final String CFG_METADATA_TOPIC_PREFIX = "samza.sql.metadataTopicPrefix";
   public static final String CFG_GROUPBY_WINDOW_DURATION_MS = "samza.sql.groupby.window.ms";
+  public static final String CFG_SQL_PROCESS_SYSTEM_EVENTS = "samza.sql.processSystemEvents";
 
   public static final String SAMZA_SYSTEM_LOG = "log";
 
@@ -115,6 +116,7 @@ public class SamzaSqlApplicationConfig {
 
   private final String metadataTopicPrefix;
   private final long windowDurationMs;
+  private final boolean processSystemEvents;
 
   public SamzaSqlApplicationConfig(Config staticConfig, List<String> inputSystemStreams,
       List<String> outputSystemStreams) {
@@ -165,6 +167,8 @@ public SamzaSqlApplicationConfig(Config staticConfig, List<String> inputSystemSt
 
     metadataTopicPrefix =
         staticConfig.get(CFG_METADATA_TOPIC_PREFIX, DEFAULT_METADATA_TOPIC_PREFIX);
+
+    processSystemEvents = staticConfig.getBoolean(CFG_SQL_PROCESS_SYSTEM_EVENTS, true);
     windowDurationMs = staticConfig.getLong(CFG_GROUPBY_WINDOW_DURATION_MS, DEFAULT_GROUPBY_WINDOW_DURATION_MS);
   }
 
@@ -324,4 +328,8 @@ public String getMetadataTopicPrefix() {
   public long getWindowDurationMs() {
     return windowDurationMs;
   }
+
+  public boolean isProcessSystemEvents() {
+    return processSystemEvents;
+  }
 }

samza-sql/src/main/java/org/apache/samza/sql/translator/ProjectTranslator.java

@@ -193,5 +193,4 @@ void translate(final Project project, final String logicalOpId, final Translator
     context.registerMessageStream(project.getId(), outputStream);
     context.registerRelNode(project.getId(), project);
   }
-
 }

samza-sql/src/main/java/org/apache/samza/sql/translator/QueryTranslator.java

@@ -50,6 +50,7 @@
 import org.apache.samza.operators.KV;
 import org.apache.samza.operators.MessageStream;
 import org.apache.samza.operators.OutputStream;
+import org.apache.samza.operators.functions.FilterFunction;
 import org.apache.samza.operators.functions.MapFunction;
 import org.apache.samza.serializers.KVSerde;
 import org.apache.samza.serializers.NoOpSerde;
@@ -293,6 +294,17 @@ private void sendToOutputStream(String queryLogicalId, String logicalOpId, Strin
       GenericOutputDescriptor<KV<Object, Object>> osd = sd.getOutputDescriptor(sinkConfig.getStreamId(), noOpKVSerde);
       OutputStream stm = outputMsgStreams.computeIfAbsent(sinkConfig.getSource(), v -> appDesc.getOutputStream(osd));
       outputStream.sendTo(stm);
+
+      // Process system events only if the output is a stream.
+      if (sqlConfig.isProcessSystemEvents()) {
+        for( MessageStream<SamzaSqlInputMessage> inputStream :  inputMsgStreams.values()) {
+          MessageStream<KV<Object, Object>> systemEventStream =
+              inputStream.filter(message -> message.getMetadata().isSystemMessage())
+                  .map(SamzaSqlInputMessage::getKeyAndMessageKV);
+
+          systemEventStream.sendTo(stm);
+        }
+      }
     } else {
       Table outputTable = appDesc.getTable(tableDescriptor.get());
       if (outputTable == null) {

samza-sql/src/main/java/org/apache/samza/sql/translator/ScanTranslator.java

@@ -1,21 +1,21 @@
 /*
-* 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.
-*/
+ * 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.translator;
 
@@ -36,8 +36,8 @@
 import org.apache.samza.operators.functions.FilterFunction;
 import org.apache.samza.operators.functions.MapFunction;
 import org.apache.samza.serializers.NoOpSerde;
-import org.apache.samza.sql.SamzaSqlInputTransformer;
 import org.apache.samza.sql.SamzaSqlInputMessage;
+import org.apache.samza.sql.SamzaSqlInputTransformer;
 import org.apache.samza.sql.data.SamzaSqlRelMessage;
 import org.apache.samza.sql.interfaces.SamzaRelConverter;
 import org.apache.samza.sql.interfaces.SqlIOConfig;
@@ -48,6 +48,7 @@
 import org.apache.samza.table.descriptors.CachingTableDescriptor;
 import org.apache.samza.table.descriptors.RemoteTableDescriptor;
 
+
 /**
  * Translator to translate the TableScans in relational graph to the corresponding input streams in the StreamGraph
  * implementation
@@ -78,7 +79,7 @@ public void init(Context context) {
 
     @Override
     public boolean apply(SamzaSqlInputMessage samzaSqlInputMessage) {
-      return !relConverter.isSystemMessage(samzaSqlInputMessage.getKeyAndMessageKV());
+      return !samzaSqlInputMessage.getMetadata().isSystemMessage();
     }
   }
 
@@ -147,11 +148,11 @@ private void updateMetrics(Instant startProcessing, Instant endProcessing) {
       queryInputEvents.inc();
       processingTime.update(Duration.between(startProcessing, endProcessing).toMillis());
     }
-
   } // ScanMapFunction
 
-  void translate(final TableScan tableScan, final String queryLogicalId, final String logicalOpId, final TranslatorContext context,
-      Map<String, DelegatingSystemDescriptor> systemDescriptors, Map<String, MessageStream<SamzaSqlInputMessage>> inputMsgStreams) {
+  void translate(final TableScan tableScan, final String queryLogicalId, final String logicalOpId,
+      final TranslatorContext context, Map<String, DelegatingSystemDescriptor> systemDescriptors,
+      Map<String, MessageStream<SamzaSqlInputMessage>> inputMsgStreams) {
     StreamApplicationDescriptor streamAppDesc = context.getStreamAppDescriptor();
     List<String> tableNameParts = tableScan.getTable().getQualifiedName();
     String sourceName = SqlIOConfig.getSourceFromSourceParts(tableNameParts);
@@ -162,9 +163,9 @@ void translate(final TableScan tableScan, final String queryLogicalId, final Str
     final String streamId = sqlIOConfig.getStreamId();
     final String source = sqlIOConfig.getSource();
 
-    final boolean isRemoteTable = sqlIOConfig.getTableDescriptor().isPresent() &&
-        (sqlIOConfig.getTableDescriptor().get() instanceof RemoteTableDescriptor ||
-            sqlIOConfig.getTableDescriptor().get() instanceof CachingTableDescriptor);
+    final boolean isRemoteTable = sqlIOConfig.getTableDescriptor().isPresent() && (
+        sqlIOConfig.getTableDescriptor().get() instanceof RemoteTableDescriptor || sqlIOConfig.getTableDescriptor()
+            .get() instanceof CachingTableDescriptor);
 
     // For remote table, we don't have an input stream descriptor. The table descriptor is already defined by the
     // SqlIOResolverFactory.
@@ -181,22 +182,55 @@ void translate(final TableScan tableScan, final String queryLogicalId, final Str
       systemDescriptors.put(systemName, systemDescriptor);
     } else {
       /* in SamzaSQL, there should be no systemDescriptor setup by user, so this branch happens only
-      * in case of Fan-OUT (i.e., same input stream used in multiple sql statements), or when same input
-      * used twice in same sql statement (e.g., select ... from input as i1, input as i2 ...), o.w., throw error */
+       * in case of Fan-OUT (i.e., same input stream used in multiple sql statements), or when same input
+       * used twice in same sql statement (e.g., select ... from input as i1, input as i2 ...), o.w., throw error */
       if (systemDescriptor.getTransformer().isPresent()) {
         InputTransformer existingTransformer = systemDescriptor.getTransformer().get();
         if (!(existingTransformer instanceof SamzaSqlInputTransformer)) {
-          throw new SamzaException("SamzaSQL Exception: existing transformer for " + systemName + " is not SamzaSqlInputTransformer");
+          throw new SamzaException(
+              "SamzaSQL Exception: existing transformer for " + systemName + " is not SamzaSqlInputTransformer");
         }
       }
     }
 
     InputDescriptor inputDescriptor = systemDescriptor.getInputDescriptor(streamId, new NoOpSerde<>());
-    MessageStream<SamzaSqlRelMessage> samzaSqlRelMessageStream =
-        inputMsgStreams.computeIfAbsent(source, v -> streamAppDesc.getInputStream(inputDescriptor))
-            .filter(new FilterSystemMessageFunction(sourceName, queryId))
-            .map(new ScanMapFunction(sourceName, queryId, queryLogicalId, logicalOpId));
+
+    if (!inputMsgStreams.containsKey(source)) {
+      MessageStream<SamzaSqlInputMessage> inputMsgStream = streamAppDesc.getInputStream(inputDescriptor);
+      inputMsgStreams.put(source, inputMsgStream.map(new SystemMessageMapperFunction(source, queryId)));
+    }
+    MessageStream<SamzaSqlRelMessage> samzaSqlRelMessageStream = inputMsgStreams.get(source)
+        .filter(new FilterSystemMessageFunction(sourceName, queryId))
+        .map(new ScanMapFunction(sourceName, queryId, queryLogicalId, logicalOpId));
 
     context.registerMessageStream(tableScan.getId(), samzaSqlRelMessageStream);
   }
+
+  /**
+   * Function that populates whether the message is a system message.
+   * TODO This should ideally be populated by the InputTransformer in future.
+   */
+  private static class SystemMessageMapperFunction implements MapFunction<SamzaSqlInputMessage, SamzaSqlInputMessage> {
+    private final String source;
+    private final int queryId;
+    private transient SamzaRelConverter relConverter;
+
+    public SystemMessageMapperFunction(String source, int queryId) {
+      this.source = source;
+      this.queryId = queryId;
+    }
+
+    @Override
+    public void init(Context context) {
+      TranslatorContext translatorContext =
+          ((SamzaSqlApplicationContext) context.getApplicationTaskContext()).getTranslatorContexts().get(queryId);
+      relConverter = translatorContext.getMsgConverter(source);
+    }
+
+    @Override
+    public SamzaSqlInputMessage apply(SamzaSqlInputMessage message) {
+      message.getMetadata().setIsSystemMessage(relConverter.isSystemMessage(message.getKeyAndMessageKV()));
+      return message;
+    }
+  }
 }

samza-sql/src/main/java/org/apache/samza/sql/translator/TranslatorInputMetricsMapFunction.java

@@ -19,7 +19,6 @@
 
 package org.apache.samza.sql.translator;
 
-import com.google.common.annotations.VisibleForTesting;
 import java.time.Instant;
 import org.apache.samza.context.ContainerContext;
 import org.apache.samza.context.Context;
@@ -60,7 +59,7 @@ public void init(Context context) {
   @Override
   public SamzaSqlRelMessage apply(SamzaSqlRelMessage message) {
     inputEvents.inc();
-    message.getSamzaSqlRelMsgMetadata().operatorBeginProcessingInstant = Instant.now().toString();
+    message.getSamzaSqlRelMsgMetadata().joinStartTimeMs = Instant.now().toEpochMilli();
     return message;
   }
 

samza-sql/src/main/java/org/apache/samza/sql/translator/TranslatorOutputMetricsMapFunction.java

@@ -63,7 +63,7 @@ public void init(Context context) {
   @Override
   public SamzaSqlRelMessage apply(SamzaSqlRelMessage message) {
     Instant endProcessing = Instant.now();
-    Instant beginProcessing = Instant.parse(message.getSamzaSqlRelMsgMetadata().operatorBeginProcessingInstant);
+    Instant beginProcessing = Instant.ofEpochMilli(message.getSamzaSqlRelMsgMetadata().joinStartTimeMs);
     outputEvents.inc();
     processingTime.update(Duration.between(beginProcessing, endProcessing).toMillis());
     return message;

samza-test/src/test/java/org/apache/samza/test/samzasql/TestSamzaSqlEndToEnd.java

@@ -103,6 +103,29 @@ public void testEndToEndWithSystemMessages() {
     staticConfigs.put(SamzaSqlApplicationConfig.CFG_SQL_STMTS_JSON, JsonUtil.toJson(sqlStmts));
     runApplication(new MapConfig(staticConfigs));
 
+    List<Integer> outMessages = TestAvroSystemFactory.messages.stream()
+        .map(x -> Integer.valueOf(((GenericRecord) x.getMessage()).get("id").toString()))
+        .sorted()
+        .collect(Collectors.toList());
+    Assert.assertEquals(numMessages, outMessages.size());
+  }
+
+  @Test
+  public void testEndToEndDisableSystemMessages() {
+    int numMessages = 20;
+
+    TestAvroSystemFactory.messages.clear();
+    Map<String, String> staticConfigs = SamzaSqlTestConfig.fetchStaticConfigsWithFactories(configs, numMessages);
+    String avroSamzaToRelMsgConverterDomain =
+        String.format(SamzaSqlApplicationConfig.CFG_FMT_SAMZA_REL_CONVERTER_DOMAIN, "avro");
+    staticConfigs.put(avroSamzaToRelMsgConverterDomain + SamzaSqlApplicationConfig.CFG_FACTORY,
+        SampleRelConverterFactory.class.getName());
+    String sql = "Insert into testavro.simpleOutputTopic select * from testavro.SIMPLE1";
+    List<String> sqlStmts = Arrays.asList(sql);
+    staticConfigs.put(SamzaSqlApplicationConfig.CFG_SQL_STMTS_JSON, JsonUtil.toJson(sqlStmts));
+    staticConfigs.put(SamzaSqlApplicationConfig.CFG_SQL_PROCESS_SYSTEM_EVENTS, "false");
+    runApplication(new MapConfig(staticConfigs));
+
     List<Integer> outMessages = TestAvroSystemFactory.messages.stream()
         .map(x -> Integer.valueOf(((GenericRecord) x.getMessage()).get("id").toString()))
         .sorted()
@@ -174,7 +197,7 @@ public void testEndToEndMultiSqlStmts() {
     Assert.assertEquals(numMessages, outMessagesSet.size());
     Assert.assertTrue(IntStream.range(0, numMessages).boxed().collect(Collectors.toList()).equals(new ArrayList<>(outMessagesSet)));
   }
-  
+
   @Test
   public void testEndToEndMultiSqlStmtsWithSameSystemStreamAsInputAndOutput() {
     int numMessages = 20;