[apex-malhar] 02/02: APEXMALHAR-2542 simplify test for fast embedded execution.

[th...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

thw pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/apex-malhar.git

commit 8e43bacba99f89a744498d6112b4853de112de9c
Author: Thomas Weise <th...@apache.org>
AuthorDate: Sat Sep 30 14:22:56 2017 -0700

    APEXMALHAR-2542 simplify test for fast embedded execution.
---
 .../examples/enricher/EnricherAppWithJSONFile.java | 32 ++++++---
 .../apex/examples/enricher/ApplicationTest.java    | 81 ++++++++++------------
 .../apex/malhar/lib/function/FunctionOperator.java |  9 +++
 3 files changed, 68 insertions(+), 54 deletions(-)

diff --git a/examples/enricher/src/main/java/org/apache/apex/examples/enricher/EnricherAppWithJSONFile.java b/examples/enricher/src/main/java/org/apache/apex/examples/enricher/EnricherAppWithJSONFile.java
index 25330e2..89a7487 100644
--- a/examples/enricher/src/main/java/org/apache/apex/examples/enricher/EnricherAppWithJSONFile.java
+++ b/examples/enricher/src/main/java/org/apache/apex/examples/enricher/EnricherAppWithJSONFile.java
@@ -19,10 +19,15 @@
 
 package org.apache.apex.examples.enricher;
 
+import java.io.PrintStream;
 import java.util.ArrayList;
 
+import org.apache.apex.malhar.lib.function.Function;
+import org.apache.apex.malhar.lib.function.FunctionOperator.MapFunctionOperator;
 import org.apache.hadoop.conf.Configuration;
 
+import com.google.common.annotations.VisibleForTesting;
+
 import com.datatorrent.api.DAG;
 import com.datatorrent.api.StreamingApplication;
 import com.datatorrent.api.annotation.ApplicationAnnotation;
@@ -36,9 +41,20 @@ import com.datatorrent.contrib.parser.JsonParser;
  */
 public class EnricherAppWithJSONFile implements StreamingApplication
 {
-
-
-
+  @SuppressWarnings("serial")
+  public static final Function.MapFunction<Object, Void> CONSOLE_OUTPUT_FN = new Function.MapFunction<Object, Void>()
+  {
+    @Override
+    public Void f(Object input)
+    {
+      PrintStream out = System.out;
+      out.println(input);
+      return null;
+    }
+  };
+
+  @VisibleForTesting
+  Function.MapFunction<Object, ?> outputFn = CONSOLE_OUTPUT_FN;
   DataGenerator dataGenerator;
 
   @Override
@@ -56,20 +72,18 @@ public class EnricherAppWithJSONFile implements StreamingApplication
     POJOEnricher enrich = dag.addOperator("Enrich", POJOEnricher.class);
     enrich.setStore(fsLoader);
 
-    ArrayList includeFields = new ArrayList();
+    ArrayList<String> includeFields = new ArrayList<>();
     includeFields.add("circleName");
-    ArrayList lookupFields = new ArrayList();
+    ArrayList<String> lookupFields = new ArrayList<>();
     lookupFields.add("circleId");
 
     enrich.setIncludeFields(includeFields);
     enrich.setLookupFields(lookupFields);
 
-    LineOutputOperator fileOut = dag.addOperator("fileOut", LineOutputOperator.class);
-
-
+    MapFunctionOperator<Object, ?> out = dag.addOperator("out", new MapFunctionOperator<>(outputFn));
     dag.addStream("Parse", dataGenerator.output, parser.in);
     dag.addStream("Enrich", parser.out, enrich.input);
-    dag.addStream("Console", enrich.output, fileOut.input);
+    dag.addStream("Console", enrich.output, out.input);
   }
 
   public DataGenerator getDataGenerator()
diff --git a/examples/enricher/src/test/java/org/apache/apex/examples/enricher/ApplicationTest.java b/examples/enricher/src/test/java/org/apache/apex/examples/enricher/ApplicationTest.java
index 8f1aed3..6b5017b 100644
--- a/examples/enricher/src/test/java/org/apache/apex/examples/enricher/ApplicationTest.java
+++ b/examples/enricher/src/test/java/org/apache/apex/examples/enricher/ApplicationTest.java
@@ -18,29 +18,43 @@
  */
 package org.apache.apex.examples.enricher;
 
-import java.io.BufferedReader;
-import java.io.File;
-import java.io.FileReader;
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import org.codehaus.jackson.map.ObjectMapper;
 import org.junit.Test;
 
 import org.apache.apex.api.EmbeddedAppLauncher;
 import org.apache.apex.api.Launcher;
+import org.apache.apex.api.Launcher.ShutdownMode;
+import org.apache.apex.malhar.lib.function.Function;
 import org.apache.hadoop.conf.Configuration;
 
 import com.datatorrent.api.Attribute;
 
 import static junit.framework.Assert.assertEquals;
 import static junit.framework.Assert.assertNotNull;
+import static junit.framework.Assert.assertTrue;
 
 /**
  * Test the DAG declaration in local mode.
  */
 public class ApplicationTest
 {
+  private static final List<Object> results = Collections.synchronizedList(new ArrayList<>());
+  @SuppressWarnings("serial")
+  private static final Function.MapFunction<Object, Void> outputFn = new Function.MapFunction<Object, Void>()
+  {
+    @Override
+    public Void f(Object input)
+    {
+      results.add(input);
+      return null;
+    }
+  };
 
   @Test
   public void testApplication() throws Exception
@@ -51,64 +65,41 @@ public class ApplicationTest
     Attribute.AttributeMap launchAttributes = new Attribute.AttributeMap.DefaultAttributeMap();
     launchAttributes.put(EmbeddedAppLauncher.RUN_ASYNC, true);
     EnricherAppWithJSONFile enricherAppWithJSONFile = new EnricherAppWithJSONFile();
+    enricherAppWithJSONFile.outputFn = outputFn;
     Launcher.AppHandle appHandle = launcher.launchApp(enricherAppWithJSONFile, conf, launchAttributes);
-    File folder = new File("target/enrichedData");
-    File[] listOfFiles = null;
     int sleepTimeCounterForLoopExit = 0;
-    int sleepTimePerIteration = 5000;
-    // First ensure that the output file is generated and the expected lines match
-    // The test main thread sleeps until the expected lines are generated in the output file
-    while (true) {
+    int sleepTimePerIteration = 500;
+    // wait until expected result count or timeout
+    while (results.size() < enricherAppWithJSONFile.getDataGenerator().getLimit()) {
       sleepTimeCounterForLoopExit += sleepTimePerIteration;
-      // break the loop of the test output is not generated by 40 seconds
-      if (sleepTimeCounterForLoopExit > 40000) {
+      if (sleepTimeCounterForLoopExit > 30000) {
         break;
       }
       Thread.sleep(sleepTimePerIteration);
-      listOfFiles = folder.listFiles();
-      if ( listOfFiles.length == 0) {
-        continue;
-      }
-      String lineRead = null;
-      int counter = 0;
-      try (BufferedReader reader = new BufferedReader(new FileReader(listOfFiles[0].getAbsolutePath()))) {
-        while ((lineRead = reader.readLine()) != null) {
-          counter += 1;
-        }
-      }
-      if (counter < enricherAppWithJSONFile.getDataGenerator().getLimit()) {
-        continue;
-      } else {
-        break;
-      }
     }
-    listOfFiles = folder.listFiles();
-    assertEnrichedOutput(listOfFiles);
+    appHandle.shutdown(ShutdownMode.KILL);
+    assertTrue(results.size() >= enricherAppWithJSONFile.getDataGenerator().getLimit());
+    assertEnrichedOutput();
   }
 
-  private void assertEnrichedOutput(File[] listOfFiles) throws IOException
+  private void assertEnrichedOutput() throws IOException
   {
-    // Ensure that only one file is generated as output per properties configuration
-    assertEquals(1, listOfFiles.length);
-    String aJSONEntry = null;
     ObjectMapper mapper = new ObjectMapper();
     Map<String,Integer> countsOfEnrichedValues = new HashMap<>();
     String enrichedValue = null;
-    try (BufferedReader reader = new BufferedReader(new FileReader(listOfFiles[0].getAbsolutePath()))) {
-      while ((aJSONEntry = reader.readLine()) != null) {
-        POJOEnriched anEnrichedEntry = mapper.readValue(aJSONEntry, POJOEnriched.class);
-        enrichedValue = anEnrichedEntry.getCircleName();
-        assertNotNull(enrichedValue); // all values should have an entiched value
-        if (!countsOfEnrichedValues.containsKey(enrichedValue)) {
-          countsOfEnrichedValues.put(enrichedValue,1);
-        } else {
-          Integer currentCountForEnrichedEntry = countsOfEnrichedValues.get(enrichedValue);
-          countsOfEnrichedValues.put(enrichedValue, currentCountForEnrichedEntry + 1);
-        }
+    for (Object result : results) {
+      POJOEnriched anEnrichedEntry = mapper.readValue(result.toString(), POJOEnriched.class);
+      enrichedValue = anEnrichedEntry.getCircleName();
+      assertNotNull(enrichedValue); // all values should have an entiched value
+      if (!countsOfEnrichedValues.containsKey(enrichedValue)) {
+        countsOfEnrichedValues.put(enrichedValue,1);
+      } else {
+        Integer currentCountForEnrichedEntry = countsOfEnrichedValues.get(enrichedValue);
+        countsOfEnrichedValues.put(enrichedValue, currentCountForEnrichedEntry + 1);
       }
     }
     for (String circleName : countsOfEnrichedValues.keySet() ) {
-      int currentCircleNameAsciiCode = ((int)circleName.charAt(0));
+      int currentCircleNameAsciiCode = (circleName.charAt(0));
       Integer circleNameCount = countsOfEnrichedValues.get(circleName);
       // All circle names  "A" to "E" should have a count of 2 and all others should have 1
       if ( ( currentCircleNameAsciiCode > 64 ) && (currentCircleNameAsciiCode < 70) ) {
diff --git a/library/src/main/java/org/apache/apex/malhar/lib/function/FunctionOperator.java b/library/src/main/java/org/apache/apex/malhar/lib/function/FunctionOperator.java
index 7b06cbf..6c6a8b4 100644
--- a/library/src/main/java/org/apache/apex/malhar/lib/function/FunctionOperator.java
+++ b/library/src/main/java/org/apache/apex/malhar/lib/function/FunctionOperator.java
@@ -79,7 +79,16 @@ public class FunctionOperator<OUT, FUNCTION extends Function> implements Operato
   {
     isAnnonymous = f.getClass().isAnonymousClass();
     if (isAnnonymous) {
+      // legacy behavior for functions that don't support Java serialization
       annonymousFunctionClass = functionClassData(f);
+      try {
+        readFunction();
+      } catch (Exception e) {
+        // cannot be handled by custom serialization, but may be Java serializable
+        annonymousFunctionClass = null;
+        statefulF.setValue(f);
+        stateful = true;
+      }
     } else if (f instanceof Function.Stateful) {
       statelessF = f;
     } else {

-- 
To stop receiving notification emails like this one, please contact
"commits@apex.apache.org" <co...@apex.apache.org>.