HERE Data SDK for Java & Scala Release 2.19.0

Author: heremaps-bot

data-archive/java/avro-example/README.md

@@ -3,7 +3,7 @@
 
     <groupId>com.here.platform.data.archive</groupId>
     <artifactId>data-archive-avro-example</artifactId>
-    <version>0.0.548</version>
+    <version>0.0.565</version>
     <packaging>jar</packaging>
 
     <!-- Meta information section start -->
@@ -23,13 +23,14 @@
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <java.version>1.8</java.version>
 
-        <sdk-bom.version>2.18.5</sdk-bom.version>
+        <sdk-bom.version>2.19.10</sdk-bom.version>
         <scala.compat.version>2.11</scala.compat.version>
 
         <avro.version>1.10.0</avro.version>
         <commons-logging.version>1.2</commons-logging.version>
 
         <junit.version>4.12</junit.version>
+        <mockito-all.version>1.10.19</mockito-all.version>
     </properties>
 
     <dependencyManagement>
@@ -45,6 +46,20 @@
         </dependencies>
     </dependencyManagement>
     <dependencies>
+        <!-- Apache Flink dependencies -->
+        <!-- These dependencies are provided, because they should not be packaged into the JAR file. -->
+        <!-- Versions for these dependencies come from environment-stream pom -->
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-core</artifactId>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-metrics-core</artifactId>
+            <scope>provided</scope>
+        </dependency>
+
         <!-- External Dependencies from sdk-dep-common-2.0.0 pom -->
         <dependency>
             <groupId>org.apache.avro</groupId>
@@ -92,6 +107,12 @@
             <version>${junit.version}</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-all</artifactId>
+            <version>${mockito-all.version}</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>
@@ -172,11 +193,21 @@
                 </property>
             </activation>
             <dependencies>
+                <dependency>
+                    <groupId>org.apache.flink</groupId>
+                    <artifactId>flink-core</artifactId>
+                    <scope>compile</scope>
+                </dependency>
                 <dependency>
                     <groupId>org.apache.flink</groupId>
                     <artifactId>flink-java</artifactId>
                     <scope>compile</scope>
                 </dependency>
+                <dependency>
+                    <groupId>org.apache.flink</groupId>
+                    <artifactId>flink-metrics-core</artifactId>
+                    <scope>compile</scope>
+                </dependency>
                 <dependency>
                     <groupId>org.apache.flink</groupId>
                     <artifactId>flink-streaming-java_${scala.compat.version}</artifactId>

data-archive/java/avro-example/pom.xml

@@ -34,6 +34,9 @@
 import java.util.Spliterators;
 import java.util.TreeSet;
 import java.util.stream.StreamSupport;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.metrics.Counter;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -45,9 +48,18 @@ public class AvroSimpleKeyExample implements SimpleUDF {
   static final String INGESTION_TIME = "ingestionTime";
   static final String EVENT_TYPE = "eventType";
   static final int ZOOM_LEVEL = 8;
+  private Counter getKeysDuration;
+  private Counter aggregateDuration;
+
+  @Override
+  public void open(Configuration parameters, RuntimeContext runtimeContext) {
+    getKeysDuration = runtimeContext.getMetricGroup().counter("getKeysDuration");
+    aggregateDuration = runtimeContext.getMetricGroup().counter("aggregateDuration");
+  }
 
   @Override
   public Map<String, Object> getKeys(Map<MetadataName, String> metadata, byte[] payload) {
+    long startTime = System.currentTimeMillis();
     SdiiMessage.Message sdiiMessage;
     try {
       sdiiMessage = SdiiMessage.Message.parseFrom(payload);
@@ -93,11 +105,13 @@ public Map<String, Object> getKeys(Map<MetadataName, String> metadata, byte[] pa
       result.put(EVENT_TYPE, event);
     }
 
+    getKeysDuration.inc(System.currentTimeMillis() - startTime);
     return result;
   }
 
   @Override
   public byte[] aggregate(Map<String, Object> keys, Iterator<byte[]> messages) {
+    long startTime = System.currentTimeMillis();
     try {
       Iterator<SdiiMessage.Message> sdiiMessages =
           StreamSupport.stream(
@@ -114,6 +128,8 @@ public byte[] aggregate(Map<String, Object> keys, Iterator<byte[]> messages) {
       return AvroHelper.aggregateProtobufMessagesAsAvro(sdiiMessages, SdiiMessage.Message.class);
     } catch (Exception e) {
       LOG.error("Aggregation errors....", e);
+    } finally {
+      aggregateDuration.inc(System.currentTimeMillis() - startTime);
     }
     return null;
   }

data-archive/java/avro-example/src/main/java/com/here/platform/data/archive/example/AvroSimpleKeyExample.java

@@ -20,6 +20,11 @@
 package com.here.platform.data.archive.example;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 
 import com.here.olp.util.quad.factory.HereQuadFactory;
 import com.here.platform.dal.custom.MetadataName;
@@ -37,11 +42,44 @@
 import java.util.Map;
 import java.util.TimeZone;
 import org.apache.commons.io.FileUtils;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.SimpleCounter;
+import org.junit.Before;
 import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.runners.MockitoJUnitRunner;
 
+@RunWith(MockitoJUnitRunner.class)
 public class AvroSimpleKeyExampleTest {
 
-  private AvroSimpleKeyExample example = new AvroSimpleKeyExample();
+  @Mock Configuration parameters;
+  @Mock RuntimeContext runtimeContext;
+  @Mock MetricGroup metricGroup;
+
+  private AvroSimpleKeyExample example;
+  private Counter getKeysDuration = new SimpleCounter();
+  private Counter aggregateDuration = new SimpleCounter();
+
+  @Before
+  public void init() {
+    when(runtimeContext.getMetricGroup()).thenReturn(metricGroup);
+    when(metricGroup.counter("getKeysDuration")).thenReturn(getKeysDuration);
+    when(metricGroup.counter("aggregateDuration")).thenReturn(aggregateDuration);
+    example = new AvroSimpleKeyExample();
+  }
+
+  @Test
+  public void testRegisterMetrics() {
+    example.open(parameters, runtimeContext);
+
+    verify(runtimeContext, times(2)).getMetricGroup();
+    verify(metricGroup, times(1)).counter(eq("getKeysDuration"));
+    verify(metricGroup, times(1)).counter(eq("aggregateDuration"));
+  }
 
   @Test
   public void testGetKeys() throws IOException {
@@ -51,10 +89,12 @@ public void testGetKeys() throws IOException {
     double latitude = 10d;
     Calendar cal = Calendar.getInstance(TimeZone.getTimeZone("UTC"));
     cal.setTimeInMillis(timestamp);
+    example.open(parameters, runtimeContext);
 
     SdiiMessage.Message sdiiMessage = prepareSDIIMessage(timestamp, true, longitude, latitude);
     Map<MetadataName, String> metadata = new HashMap<>();
     metadata.put(MetadataName.INGESTION_TIME, String.valueOf(timestamp));
+    assertEquals(0, getKeysDuration.getCount());
     Map<String, Object> keys = example.getKeys(metadata, sdiiMessage.toByteArray());
     assertEquals(
         HereQuadFactory.INSTANCE
@@ -65,6 +105,7 @@ public void testGetKeys() throws IOException {
     assertEquals(
         SdiiCommon.SignRecognition.getDescriptor().getName(),
         keys.get(AvroSimpleKeyExample.EVENT_TYPE));
+    assertTrue(getKeysDuration.getCount() > 0);
   }
 
   @Test
@@ -74,7 +115,9 @@ public void testAggregate() throws IOException {
     SdiiMessage.Message message3 = prepareSDIIMessage(System.currentTimeMillis(), true, 10d, 10d);
     List<byte[]> messagesList =
         Arrays.asList(message1.toByteArray(), message2.toByteArray(), message3.toByteArray());
+    example.open(parameters, runtimeContext);
 
+    assertEquals(0, aggregateDuration.getCount());
     File tmpFile = File.createTempFile("test", ".avro");
     tmpFile.deleteOnExit();
     FileUtils.writeByteArrayToFile(
@@ -85,6 +128,7 @@ public void testAggregate() throws IOException {
     assertSDIIMessagesAreEqual(message1, list.get(0));
     assertSDIIMessagesAreEqual(message2, list.get(1));
     assertSDIIMessagesAreEqual(message3, list.get(2));
+    assertTrue(aggregateDuration.getCount() > 0);
   }
 
   @Test
@@ -93,6 +137,7 @@ public void testMerge() throws IOException {
     SdiiMessage.Message message2 = prepareSDIIMessage(System.currentTimeMillis(), true, 10d, 10d);
     SdiiMessage.Message message3 = prepareSDIIMessage(System.currentTimeMillis(), true, 10d, 10d);
     SdiiMessage.Message message4 = prepareSDIIMessage(System.currentTimeMillis(), true, 10d, 10d);
+    example.open(parameters, runtimeContext);
     byte[] archiveBytesMessage1 =
         example.aggregate(
             null, Arrays.asList(message1.toByteArray(), message2.toByteArray()).iterator());