Adding test data sources designed for benchmarking purposes

hazelcast-jet-core/src/main/java/com/hazelcast/jet/pipeline/test/StreamSourceLong.java

@@ -0,0 +1,155 @@
+/*
+ * Copyright (c) 2008-2020, Hazelcast, Inc. All Rights Reserved.
+ *
+ * Licensed 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 com.hazelcast.jet.pipeline.test;
+
+import com.hazelcast.jet.core.AbstractProcessor;
+import com.hazelcast.jet.core.AppendableTraverser;
+import com.hazelcast.jet.core.EventTimePolicy;
+import com.hazelcast.jet.core.Watermark;
+
+import javax.annotation.Nonnull;
+
+import static com.hazelcast.jet.impl.JetEvent.jetEvent;
+import static java.util.concurrent.TimeUnit.HOURS;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static java.util.concurrent.TimeUnit.NANOSECONDS;
+import static java.util.concurrent.TimeUnit.SECONDS;
+
+/**
+ * Represents a StreamSource containing Long values which can be used
+ * for performance pipeline testing and development.
+ *
+ * @since 4.3
+ */
+public class StreamSourceLong extends AbstractProcessor {
+
+    private static final long SOURCE_THROUGHPUT_REPORTING_PERIOD_SECONDS = 10;
+    private static final long SIMPLE_TIME_SPAN_MILLIS = HOURS.toMillis(3);
+
+    private static final long REPORT_PERIOD_NANOS = SECONDS.toNanos(SOURCE_THROUGHPUT_REPORTING_PERIOD_SECONDS);
+    private static final long HICCUP_REPORT_THRESHOLD_MILLIS = 10;
+    private final long nanoTimeMillisToCurrentTimeMillis = determineTimeOffset();
+    private final long startTime;
+    private final long itemsPerSecond;
+    private final boolean isReportingThroughput;
+    private final long wmGranularity;
+    private final long wmOffset;
+    private long globalProcessorIndex;
+    private long totalParallelism;
+    private long emitPeriod;
+
+    private final AppendableTraverser<Object> traverser = new AppendableTraverser<>(2);
+    private long emitSchedule;
+    private long lastReport;
+    private long counterAtLastReport;
+    private long lastCallNanos;
+    private long counter;
+    private long lastEmittedWm;
+    private long nowNanos;
+
+    /**
+     * Creates a stream source for Long values that is supposed to be used for performance testing purposes.
+     *
+     * @since 4.3
+     */
+    @Nonnull
+    public StreamSourceLong(
+            long startTime,
+            long itemsPerSecond,
+            EventTimePolicy<? super Long> eventTimePolicy,
+            boolean shouldReportThroughput
+    ) {
+        this.wmGranularity = eventTimePolicy.watermarkThrottlingFrameSize();
+        this.wmOffset = eventTimePolicy.watermarkThrottlingFrameOffset();
+        this.startTime = MILLISECONDS.toNanos(startTime + nanoTimeMillisToCurrentTimeMillis);
+        this.itemsPerSecond = itemsPerSecond;
+        this.isReportingThroughput = shouldReportThroughput;
+    }
+
+    @Override
+    protected void init(Context context) {
+        totalParallelism = context.totalParallelism();
+        globalProcessorIndex = context.globalProcessorIndex();
+        emitPeriod = SECONDS.toNanos(1) * totalParallelism / itemsPerSecond;
+        lastCallNanos = lastReport = emitSchedule =
+                startTime + SECONDS.toNanos(1) * globalProcessorIndex / itemsPerSecond;
+    }
+
+    @Override
+    public boolean complete() {
+        nowNanos = System.nanoTime();
+        detectAndReportHiccup();
+        emitEvents();
+        if (isReportingThroughput) {
+            reportThroughput();
+        }
+        return false;
+    }
+
+    private void emitEvents() {
+        while (emitFromTraverser(traverser) && emitSchedule <= nowNanos) {
+            long timestamp = NANOSECONDS.toMillis(emitSchedule) - nanoTimeMillisToCurrentTimeMillis;
+            traverser.append(jetEvent(timestamp, counter * totalParallelism + globalProcessorIndex));
+            counter++;
+            emitSchedule += emitPeriod;
+            if (timestamp >= lastEmittedWm + wmGranularity) {
+                long wmToEmit = timestamp - (timestamp % wmGranularity) + wmOffset;
+                traverser.append(new Watermark(wmToEmit));
+                lastEmittedWm = wmToEmit;
+            }
+        }
+    }
+
+    private void detectAndReportHiccup() {
+        long millisSinceLastCall = NANOSECONDS.toMillis(nowNanos - lastCallNanos);
+        if (millisSinceLastCall > HICCUP_REPORT_THRESHOLD_MILLIS) {
+            System.out.printf("*** Source #%d hiccup: %,d ms%n", globalProcessorIndex, millisSinceLastCall);
+        }
+        lastCallNanos = nowNanos;
+    }
+
+    private void reportThroughput() {
+        long nanosSinceLastReport = nowNanos - lastReport;
+        if (nanosSinceLastReport < REPORT_PERIOD_NANOS) {
+            return;
+        }
+        lastReport = nowNanos;
+        long itemCountSinceLastReport = counter - counterAtLastReport;
+        counterAtLastReport = counter;
+        System.out.printf("%,d p%d: %,.0f items/second%n",
+                simpleTime(NANOSECONDS.toMillis(nowNanos)),
+                globalProcessorIndex,
+                itemCountSinceLastReport / ((double) nanosSinceLastReport / SECONDS.toNanos(1))
+        );
+    }
+
+    @Override
+    public boolean tryProcessWatermark(Watermark watermark) {
+        throw new UnsupportedOperationException("Source processor shouldn't be asked to process a watermark");
+    }
+
+    private static long determineTimeOffset() {
+        long milliTime = System.currentTimeMillis();
+        long nanoTime = System.nanoTime();
+        return NANOSECONDS.toMillis(nanoTime) - milliTime;
+    }
+
+    private static long simpleTime(long timeMillis) {
+        return timeMillis % SIMPLE_TIME_SPAN_MILLIS;
+    }
+
+}

hazelcast-jet-core/src/main/java/com/hazelcast/jet/pipeline/test/TestSources.java

@@ -16,12 +16,17 @@
 
 package com.hazelcast.jet.pipeline.test;
 
+import com.hazelcast.cluster.Address;
 import com.hazelcast.jet.annotation.EvolvingApi;
+import com.hazelcast.jet.core.ProcessorMetaSupplier;
+import com.hazelcast.jet.core.ProcessorSupplier;
+import com.hazelcast.jet.core.Vertex;
 import com.hazelcast.jet.pipeline.BatchSource;
 import com.hazelcast.jet.pipeline.SourceBuilder;
-import com.hazelcast.jet.pipeline.SourceBuilder.TimestampedSourceBuffer;
+import com.hazelcast.jet.pipeline.Sources;
 import com.hazelcast.jet.pipeline.StreamSource;
 import com.hazelcast.jet.pipeline.StreamSourceStage;
+import com.hazelcast.jet.pipeline.SourceBuilder.TimestampedSourceBuffer;
 
 import javax.annotation.Nonnull;
 import java.util.Arrays;
@@ -161,4 +166,71 @@ void fillBuffer(TimestampedSourceBuffer<T> buf) throws Exception {
             }
         }
     }
+
+    /**
+     * Returns a stream source that contains long values and does late materialization
+     * of values after distributing them across the cluster, which is useful for high-throughput
+     * testing.
+     *
+     * @since 4.3
+     *
+     */
+    @Nonnull
+    public static StreamSource<Long> longStreamSource(long itemsPerSecond, long initialDelay) {
+        return longStreamSource(itemsPerSecond, initialDelay, Vertex.LOCAL_PARALLELISM_USE_DEFAULT, false);
+    }
+
+    /**
+     * Returns a stream source that contains long values and does late materialization
+     * of values after distributing them across the cluster, which is useful for high-throughput
+     * testing.
+     *
+     * @since 4.3
+     *
+     */
+    @Nonnull
+    public static StreamSource<Long> longStreamSource(long itemsPerSecond, long initialDelay,
+                                                      int preferredLocalParallelism) {
+        return longStreamSource(itemsPerSecond, initialDelay, preferredLocalParallelism, false);
+    }
+
+    /**
+     * Returns a stream source that contains long values and does late materialization
+     * of values after distributing them across the cluster, which is useful for high-throughput
+     * testing.
+     *
+     * @since 4.3
+     *
+     */
+    @Nonnull
+    public static StreamSource<Long> longStreamSource(long itemsPerSecond, long initialDelay,
+                                                      boolean shouldReportThroughput) {
+        return longStreamSource(itemsPerSecond, initialDelay,
+                Vertex.LOCAL_PARALLELISM_USE_DEFAULT, shouldReportThroughput);
+    }
+
+    /**
+     * Returns a stream source that contains long values and does late materialization
+     * of values after distributing them across the cluster, which is useful for high-throughput
+     * testing.
+     *
+     * @since 4.3
+     *
+     */
+    @Nonnull
+    public static StreamSource<Long> longStreamSource(long itemsPerSecond, long initialDelay,
+                                                      int preferredLocalParallelism, boolean shouldReportThroughput) {
+        return Sources.streamFromProcessorWithWatermarks("longs",
+                true,
+                eventTimePolicy -> ProcessorMetaSupplier.of(
+                        preferredLocalParallelism,
+                        (Address ignored) -> {
+                            long startTime = System.currentTimeMillis() + initialDelay;
+                            return ProcessorSupplier.of(() ->
+                                    new StreamSourceLong(startTime, itemsPerSecond,
+                                            eventTimePolicy, shouldReportThroughput));
+                        })
+        );
+
+    }
 }

hazelcast-jet-core/src/test/java/com/hazelcast/jet/pipeline/test/TestSourcesTest.java

@@ -26,6 +26,7 @@
 import java.util.Arrays;
 import java.util.List;
 import java.util.stream.IntStream;
+import java.util.stream.LongStream;
 
 import static com.hazelcast.jet.pipeline.test.Assertions.assertCollectedEventually;
 import static org.junit.Assert.assertEquals;
@@ -48,6 +49,30 @@ public void test_items() {
         jet().newJob(p).join();
     }
 
+    @Test
+    public void test_stream_source_long() throws Throwable {
+        int itemsPerSecond = 10;
+        int timeout = 10;
+        int numberOfExpectedValues = timeout * itemsPerSecond;
+        Long[] input = LongStream.range(0, numberOfExpectedValues).boxed().toArray(Long[]::new);
+        List<Long> expected = Arrays.asList(input);
+
+        p.readFrom(TestSources.longStreamSource(itemsPerSecond, 0)).
+                withIngestionTimestamps().
+                apply(assertCollectedEventually(timeout, items -> {
+                    assertTrue("list should contain at least " + numberOfExpectedValues + " items",
+                            items.size() >= numberOfExpectedValues);
+                    assertTrue("list should contain less than " + 2 * numberOfExpectedValues + " items",
+                            items.size() < 2 * numberOfExpectedValues);
+                    for (Long value : items) {
+                        assertTrue(expected.contains(value));
+                    }
+                }));
+
+        expectedException.expectMessage(AssertionCompletedException.class.getName());
+        executeAndPeel();
+    }
+
     @Test
     public void test_itemStream() throws Throwable {
         int expectedItemCount = 20;