Issue 191: Add at-least-once stream processing example

README.md

@@ -27,6 +27,7 @@ more complex applications as sub-projects, which show use-cases exploiting one o
 | `noop` | Example of how to add a simple callback executed upon a read event. | [Java](pravega-client-examples/src/main/java/io/pravega/example/noop)
 | `statesynchronizer` | Application that allows users to work with `StateSynchronizer` API via CLI. | [Java](pravega-client-examples/src/main/java/io/pravega/example/statesynchronizer)
 | `streamcuts` | Application examples demonstrating the use of `StreamCut`s via CLI. | [Java](pravega-client-examples/src/main/java/io/pravega/example/streamcuts) 
+| `streamprocessing` | An example that illustrates at-least-once processing using the Pravega API. | [Java](pravega-client-examples/src/main/java/io/pravega/example/streamprocessing)
 
 The related documentation and instructions are [here](pravega-client-examples).
 

pravega-client-examples/README.md

@@ -198,3 +198,6 @@ $ bin/secureBatchReader [-scope "myScope"] [-stream "myStream"] [-uri "tls://loc
 
 All args are optional. If not included, the default values are same as the defaults mentioned earlier for
 `bin\secureWriter`.
+
+## `streamprocessing`
+See [streamprocessing](src/main/java/io/pravega/example/streamprocessing/README.md).

pravega-client-examples/build.gradle

@@ -26,6 +26,9 @@ ext {
 
 dependencies {
     testCompile "junit:junit:${junitVersion}"
+    testCompile "io.pravega:pravega-standalone:${pravegaVersion}"
+    testCompile "io.pravega:pravega-test-testcommon:${pravegaVersion}"
+    compileOnly "org.projectlombok:lombok:${lombokVersion}"
 
     compile "io.pravega:pravega-client:${pravegaVersion}",
             "io.pravega:pravega-common:${pravegaVersion}",
@@ -182,6 +185,44 @@ task startSecureBatchReader(type: JavaExec) {
     }
 }
 
+task scriptEventGenerator(type: CreateStartScripts) {
+    outputDir = file('build/scripts')
+    mainClassName = 'io.pravega.example.streamprocessing.EventGenerator'
+    applicationName = 'eventGenerator'
+    defaultJvmOpts = ["-Dlogback.configurationFile=file:conf/logback.xml"]
+    classpath = files(jar.archivePath) + sourceSets.main.runtimeClasspath
+}
+
+task startEventGenerator(type: JavaExec) {
+    main = "io.pravega.example.streamprocessing.EventGenerator"
+    classpath = sourceSets.main.runtimeClasspath
+}
+
+task scriptAtLeastOnceApp(type: CreateStartScripts) {
+    outputDir = file('build/scripts')
+    mainClassName = 'io.pravega.example.streamprocessing.AtLeastOnceApp'
+    applicationName = 'atLeastOnceApp'
+    defaultJvmOpts = ["-Dlogback.configurationFile=file:conf/logback.xml"]
+    classpath = files(jar.archivePath) + sourceSets.main.runtimeClasspath
+}
+
+task startAtLeastOnceApp(type: JavaExec) {
+    main = "io.pravega.example.streamprocessing.AtLeastOnceApp"
+    classpath = sourceSets.main.runtimeClasspath
+}
+
+task scriptEventDebugSink(type: CreateStartScripts) {
+    outputDir = file('build/scripts')
+    mainClassName = 'io.pravega.example.streamprocessing.EventDebugSink'
+    applicationName = 'eventDebugSink'
+    defaultJvmOpts = ["-Dlogback.configurationFile=file:conf/logback.xml"]
+    classpath = files(jar.archivePath) + sourceSets.main.runtimeClasspath
+}
+
+task startEventDebugSink(type: JavaExec) {
+    main = "io.pravega.example.streamprocessing.EventDebugSink"
+    classpath = sourceSets.main.runtimeClasspath
+}
 
 distributions {
     main {
@@ -198,6 +239,9 @@ distributions {
                 from project.scriptSecureWriter
                 from project.scriptSecureReader
                 from project.scriptSecureBatchReader
+                from project.scriptEventGenerator
+                from project.scriptAtLeastOnceApp
+                from project.scriptEventDebugSink
             }
             into('lib') {
                 from(jar)

pravega-client-examples/src/main/java/io/pravega/example/streamprocessing/AppConfiguration.java

@@ -0,0 +1,94 @@
+/*
+ * Copyright (c) Dell Inc., or its subsidiaries. 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
+ *
+ */
+package io.pravega.example.streamprocessing;
+
+import java.net.URI;
+import java.util.UUID;
+
+/**
+ * All parameters will come from environment variables.
+ * This makes it easy to configure on Docker, Kubernetes, etc.
+ */
+class AppConfiguration {
+    AppConfiguration(String[] args) {
+    }
+
+    // By default, we will connect to a standalone Pravega running on localhost.
+    public URI getControllerURI() {
+        return URI.create(getEnvVar("PRAVEGA_CONTROLLER", "tcp://localhost:9090"));
+    }
+
+    public String getScope() {
+        return getEnvVar("PRAVEGA_SCOPE", "examples");
+    }
+
+    public String getInstanceId() {
+        return getEnvVar("INSTANCE_ID", UUID.randomUUID().toString());
+    }
+
+    /**
+     * The output of EventGenerator and the input of AtLeastOnceApp.
+     */
+    public String getStream1Name() {
+        return getEnvVar("PRAVEGA_STREAM_1", "streamprocessing1c");
+    }
+
+    /**
+     * The output of AtLeastOnceApp and the input of EventDebugSink.
+     */
+    public String getStream2Name() {
+        return getEnvVar("PRAVEGA_STREAM_2", "streamprocessing2c");
+    }
+
+    public String getReaderGroup() {
+        return getEnvVar("PRAVEGA_READER_GROUP", "streamprocessing1c-rg1");
+    }
+
+    public String getMembershipSynchronizerStreamName() {
+        return getReaderGroup() + "-membership";
+    }
+
+    public int getTargetRateEventsPerSec() {
+        return Integer.parseInt(getEnvVar("PRAVEGA_TARGET_RATE_EVENTS_PER_SEC", "10"));
+    }
+
+    public int getScaleFactor() {
+        return Integer.parseInt(getEnvVar("PRAVEGA_SCALE_FACTOR", "2"));
+    }
+
+    public int getMinNumSegments() {
+        return Integer.parseInt(getEnvVar("PRAVEGA_MIN_NUM_SEGMENTS", "6"));
+    }
+
+    public  long getCheckpointPeriodMs() {
+        return Long.parseLong(getEnvVar("CHECKPOINT_PERIOD_MS", "3000"));
+    }
+
+    public long getCheckpointTimeoutMs() {
+        return Long.parseLong(getEnvVar("CHECKPOINT_TIMEOUT_MS", "120000"));
+    }
+
+    public long getTransactionTimeoutMs() {
+        return Long.parseLong(getEnvVar("TRANSACTION_TIMEOUT_MS", "120000"));
+    }
+
+    public long getHeartbeatIntervalMillis() {
+        return Long.parseLong(getEnvVar("HEARTBEAT_INTERVAL_MS", "500"));
+    }
+
+    private static String getEnvVar(String name, String defaultValue) {
+        String value = System.getenv(name);
+        if (value == null || value.isEmpty()) {
+            return defaultValue;
+        }
+        return value;
+    }
+}

pravega-client-examples/src/main/java/io/pravega/example/streamprocessing/AtLeastOnceApp.java

@@ -0,0 +1,163 @@
+/*
+ * Copyright (c) Dell Inc., or its subsidiaries. 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
+ *
+ */
+package io.pravega.example.streamprocessing;
+
+import com.google.gson.reflect.TypeToken;
+import io.pravega.client.ClientConfig;
+import io.pravega.client.EventStreamClientFactory;
+import io.pravega.client.SynchronizerClientFactory;
+import io.pravega.client.admin.ReaderGroupManager;
+import io.pravega.client.admin.StreamManager;
+import io.pravega.client.stream.EventRead;
+import io.pravega.client.stream.EventStreamWriter;
+import io.pravega.client.stream.EventWriterConfig;
+import io.pravega.client.stream.ReaderConfig;
+import io.pravega.client.stream.ReaderGroup;
+import io.pravega.client.stream.ReaderGroupConfig;
+import io.pravega.client.stream.ScalingPolicy;
+import io.pravega.client.stream.Stream;
+import io.pravega.client.stream.StreamConfiguration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.Executors;
+
+/**
+ * This demonstrates reading events from a Pravega stream, processing each event,
+ * and writing each output event to another Pravega stream.
+ * It guarantees that each event is processed at least once.
+ * If multiple instances of this application are executed using the same readerGroupName parameter,
+ * each instance will get a distinct subset of events.
+ *
+ * Use {@link EventGenerator} to generate input events and {@link EventDebugSink}
+ * to view the output events.
+ */
+public class AtLeastOnceApp {
+    private static final Logger log = LoggerFactory.getLogger(AtLeastOnceApp.class);
+
+    private final AppConfiguration config;
+
+    public static void main(String[] args) throws Exception {
+        final AtLeastOnceApp app = new AtLeastOnceApp(new AppConfiguration(args));
+        app.run();
+    }
+
+    public AtLeastOnceApp(AppConfiguration config) {
+        this.config = config;
+    }
+
+    public AppConfiguration getConfig() {
+        return config;
+    }
+
+    public void run() throws Exception {
+        // Get the provided instanceId that uniquely identifes this instances of AtLeastOnceApp.
+        // It will be randomly generated if not provided by the user.
+        final String instanceId = getConfig().getInstanceId();
+        log.info("instanceId={}", instanceId);
+
+        // Define configuration to connect to Pravega.
+        final ClientConfig clientConfig = ClientConfig.builder().controllerURI(getConfig().getControllerURI()).build();
+
+        // Create the input and output streams (ignored if they already exist).
+        try (StreamManager streamManager = StreamManager.create(clientConfig)) {
+            streamManager.createScope(getConfig().getScope());
+            final StreamConfiguration streamConfig = StreamConfiguration.builder()
+                    .scalingPolicy(ScalingPolicy.byEventRate(
+                            getConfig().getTargetRateEventsPerSec(),
+                            getConfig().getScaleFactor(),
+                            getConfig().getMinNumSegments()))
+                    .build();
+            streamManager.createStream(getConfig().getScope(), getConfig().getStream1Name(), streamConfig);
+            streamManager.createStream(getConfig().getScope(), getConfig().getStream2Name(), streamConfig);
+            // Create stream for the membership state synchronizer.
+            streamManager.createStream(
+                    getConfig().getScope(),
+                    getConfig().getMembershipSynchronizerStreamName(),
+                    StreamConfiguration.builder().scalingPolicy(ScalingPolicy.fixed(1)).build());
+        }
+
+        final ReaderGroupConfig readerGroupConfig = ReaderGroupConfig.builder()
+                .stream(Stream.of(getConfig().getScope(), getConfig().getStream1Name()))
+                .automaticCheckpointIntervalMillis(getConfig().getCheckpointPeriodMs())
+                .build();
+        try (ReaderGroupManager readerGroupManager = ReaderGroupManager.withScope(getConfig().getScope(), clientConfig)) {
+            // Create the Reader Group (ignored if it already exists)
+            readerGroupManager.createReaderGroup(getConfig().getReaderGroup(), readerGroupConfig);
+            final ReaderGroup readerGroup = readerGroupManager.getReaderGroup(getConfig().getReaderGroup());
+            try (EventStreamClientFactory eventStreamClientFactory = EventStreamClientFactory.withScope(getConfig().getScope(), clientConfig);
+                 SynchronizerClientFactory synchronizerClientFactory = SynchronizerClientFactory.withScope(getConfig().getScope(), clientConfig);
+                 // Create a Pravega stream writer that we will send our processed output to.
+                 EventStreamWriter<SampleEvent> writer = eventStreamClientFactory.createEventWriter(
+                         getConfig().getStream2Name(),
+                         new JSONSerializer<>(new TypeToken<SampleEvent>(){}.getType()),
+                         EventWriterConfig.builder().build())) {
+
+                final AtLeastOnceProcessor<SampleEvent> processor = new AtLeastOnceProcessor<SampleEvent>(
+                        () -> ReaderGroupPruner.create(
+                                readerGroup,
+                                getConfig().getMembershipSynchronizerStreamName(),
+                                instanceId,
+                                synchronizerClientFactory,
+                                Executors.newScheduledThreadPool(1),
+                                getConfig().getHeartbeatIntervalMillis()),
+                        () -> eventStreamClientFactory.<SampleEvent>createReader(
+                                instanceId,
+                                readerGroup.getGroupName(),
+                                new JSONSerializer<>(new TypeToken<SampleEvent>(){}.getType()),
+                                ReaderConfig.builder().build()),
+                        1000)
+                {
+                    /**
+                     * Process an event that was read.
+                     * Processing can be performed asynchronously after this method returns.
+                     * This method must be stateless.
+                     *
+                     * For this demonstration, we output the same event that was read but with
+                     * the processedBy field set.
+                     *
+                     * @param eventRead The event read.
+                     */
+                    @Override
+                    public void process(EventRead<SampleEvent> eventRead) {
+                        final SampleEvent event = eventRead.getEvent();
+                        event.processedBy = instanceId;
+                        event.processedLatencyMs = System.currentTimeMillis() - event.timestamp;
+                        log.info("{}", event);
+                        writer.writeEvent(event.routingKey, event);
+                    }
+
+                    /**
+                     * If {@link #process} did not completely process prior events, it must do so before returning.
+                     * If writing to a Pravega stream, this should call {@link EventStreamWriter#flush}.
+                     */
+                    @Override
+                    public void flush() {
+                        writer.flush();
+                    }
+                };
+
+                processor.startAsync();
+
+                // Add shutdown hook for graceful shutdown.
+                Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+                    log.info("Running shutdown hook.");
+                    processor.stopAsync();
+                    log.info("Waiting for processor to terminate.");
+                    processor.awaitTerminated();
+                    log.info("Processor terminated.");
+                }));
+
+                processor.awaitTerminated();
+            }
+        }
+    }
+}