Merge branch 'master' of https://github.com/apache/samza

Author: sborya

build.gradle

@@ -192,16 +192,27 @@ project(':samza-azure') {
 
   dependencies {
     compile "com.microsoft.azure:azure-storage:5.3.1"
+    compile "com.microsoft.azure:azure-eventhubs:0.14.5"
     compile "com.fasterxml.jackson.core:jackson-core:2.8.8"
+    compile "io.dropwizard.metrics:metrics-core:3.1.2"
     compile project(':samza-api')
     compile project(":samza-core_$scalaVersion")
     compile "org.slf4j:slf4j-api:$slf4jVersion"
     testCompile "junit:junit:$junitVersion"
+    testCompile "org.mockito:mockito-all:$mockitoVersion"
+    testCompile "org.powermock:powermock-api-mockito:$powerMockVersion"
+    testCompile "org.powermock:powermock-core:$powerMockVersion"
+    testCompile "org.powermock:powermock-module-junit4:$powerMockVersion"
   }
   checkstyle {
     configFile = new File(rootDir, "checkstyle/checkstyle.xml")
     toolVersion = "$checkstyleVersion"
   }
+  test {
+    // Exclude integration tests that require connection to EventHub
+    exclude 'org/apache/samza/system/eventhub/producer/*ITest*'
+    exclude 'org/apache/samza/system/eventhub/consumer/*ITest*'
+  }
 }
 
 

samza-api/src/main/java/org/apache/samza/operators/MessageStream.java

@@ -73,7 +73,7 @@ public interface MessageStream<M> {
    * should be retained in the filtered {@link MessageStream}.
    *
    * @param filterFn the predicate to filter messages from this {@link MessageStream}.
-   * @return the transformed {@link MessageStream}
+   * @return the filtered {@link MessageStream}
    */
   MessageStream<M> filter(FilterFunction<? super M> filterFn);
 
@@ -105,15 +105,19 @@ public interface MessageStream<M> {
    * <p>
    * Use the {@link org.apache.samza.operators.windows.Windows} helper methods to create the appropriate windows.
    * <p>
-   * <b>Warning:</b> As of version 0.13.0, messages in windows are kept in memory and will be lost during restarts.
+   * The {@code id} must be unique for each operator in this application. It is used as part of the unique ID
+   * for any state stores and streams created by this operator (the full ID also contains the job name, job id and
+   * operator type). If the application logic is changed, this ID must be reused in the new operator to retain
+   * state from the previous version, and changed for the new operator to discard the state from the previous version.
    *
    * @param window the window to group and process messages from this {@link MessageStream}
+   * @param id the unique id of this operator in this application
    * @param <K> the type of key in the message in this {@link MessageStream}. If a key is specified,
    *            panes are emitted per-key.
    * @param <WV> the type of value in the {@link WindowPane} in the transformed {@link MessageStream}
-   * @return the transformed {@link MessageStream}
+   * @return the windowed {@link MessageStream}
    */
-  <K, WV> MessageStream<WindowPane<K, WV>> window(Window<M, K, WV> window);
+  <K, WV> MessageStream<WindowPane<K, WV>> window(Window<M, K, WV> window, String id);
 
   /**
    * Joins this {@link MessageStream} with another {@link MessageStream} using the provided
@@ -124,22 +128,27 @@ public interface MessageStream<M> {
    * <p>
    * Both inputs being joined must have the same number of partitions, and should be partitioned by the join key.
    * <p>
-   * <b>Warning:</b> As of version 0.13.0, messages in joins are kept in memory and will be lost during restarts.
+   * The {@code id} must be unique for each operator in this application. It is used as part of the unique ID
+   * for any state stores and streams created by this operator (the full ID also contains the job name, job id and
+   * operator type). If the application logic is changed, this ID must be reused in the new operator to retain
+   * state from the previous version, and changed for the new operator to discard the state from the previous version.
    *
    * @param otherStream the other {@link MessageStream} to be joined with
    * @param joinFn the function to join messages from this and the other {@link MessageStream}
-   * @param ttl the ttl for messages in each stream
    * @param keySerde the serde for the join key
    * @param messageSerde the serde for messages in this stream
    * @param otherMessageSerde the serde for messages in the other stream
+   * @param ttl the ttl for messages in each stream
+   * @param id the unique id of this operator in this application
    * @param <K> the type of join key
    * @param <OM> the type of messages in the other stream
    * @param <JM> the type of messages resulting from the {@code joinFn}
    * @return the joined {@link MessageStream}
    */
   <K, OM, JM> MessageStream<JM> join(MessageStream<OM> otherStream,
       JoinFunction<? extends K, ? super M, ? super OM, ? extends JM> joinFn,
-      Serde<K> keySerde, Serde<M> messageSerde, Serde<OM> otherMessageSerde, Duration ttl);
+      Serde<K> keySerde, Serde<M> messageSerde, Serde<OM> otherMessageSerde,
+      Duration ttl, String id);
 
   /**
    * Merges all {@code otherStreams} with this {@link MessageStream}.
@@ -186,26 +195,34 @@ static <T> MessageStream<T> mergeAll(Collection<? extends MessageStream<? extend
    * configuration, if present.
    * Else, the number of partitions is set to to the max of number of partitions for all input and output streams
    * (excluding intermediate streams).
+   * <p>
+   * The {@code id} must be unique for each operator in this application. It is used as part of the unique ID
+   * for any state stores and streams created by this operator (the full ID also contains the job name, job id and
+   * operator type). If the application logic is changed, this ID must be reused in the new operator to retain
+   * state from the previous version, and changed for the new operator to discard the state from the previous version.
    *
-   * @param <K> the type of output key
-   * @param <V> the type of output value
    * @param keyExtractor the {@link Function} to extract the message and partition key from the input message
    * @param valueExtractor the {@link Function} to extract the value from the input message
    * @param serde the {@link KVSerde} to use for (de)serializing the key and value.
+   * @param id the unique id of this operator in this application
+   * @param <K> the type of output key
+   * @param <V> the type of output value
    * @return the repartitioned {@link MessageStream}
    */
   <K, V> MessageStream<KV<K, V>> partitionBy(Function<? super M, ? extends K> keyExtractor,
-      Function<? super M, ? extends V> valueExtractor, KVSerde<K, V> serde);
+      Function<? super M, ? extends V> valueExtractor, KVSerde<K, V> serde, String id);
+
 
   /**
-   * Same as calling {@link #partitionBy(Function, Function, KVSerde)} with a null KVSerde.
+   * Same as calling {@link #partitionBy(Function, Function, KVSerde, String)} with a null KVSerde.
    *
    * @param keyExtractor the {@link Function} to extract the message and partition key from the input message
    * @param valueExtractor the {@link Function} to extract the value from the input message
+   * @param id the unique id of this operator in this application
    * @param <K> the type of output key
    * @param <V> the type of output value
    * @return the repartitioned {@link MessageStream}
    */
   <K, V> MessageStream<KV<K, V>> partitionBy(Function<? super M, ? extends K> keyExtractor,
-      Function<? super M, ? extends V> valueExtractor);
+      Function<? super M, ? extends V> valueExtractor, String id);
 }

samza-azure/src/main/java/org/apache/samza/system/eventhub/EventHubClientManager.java

@@ -0,0 +1,69 @@
+/*
+* 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.system.eventhub;
+
+import com.microsoft.azure.eventhubs.EventHubClient;
+
+/**
+ * <p>
+ * EventHubClient manager is the interface that must be implemented to wrap the
+ * {@link EventHubClient} with lifecycle hooks for initialization and close.
+ * </p>
+ *
+ * <p>
+ * {@link #init()} should be invoked once during the startup and provides a
+ * hook to perform some initialization before the creation of the underlying
+ * {@link EventHubClient}. {@link #close(long)} is invoked once during shut-down
+ * and can be used to perform clean-ups.
+ * </p>
+ */
+public interface EventHubClientManager {
+  /**
+   * A constant that can be used in the close method's timeout parameter to
+   * denote that the close invocation should block until all the teardown
+   * operations for the {@link EventHubClient} are completed
+   */
+  public static int BLOCK_UNTIL_CLOSE = -1;
+
+  /**
+   * Lifecycle hook to perform initializations for the creation of
+   * the underlying {@link EventHubClient}.
+   */
+  void init();
+
+  /**
+   * Returns the underlying {@link EventHubClient} instance. Multiple invocations
+   * of this method should return the same instance instead of
+   * creating new ones.
+   *
+   * @return EventHub client instance of the wrapper
+   */
+  EventHubClient getEventHubClient();
+
+  /**
+   * Tries to close the {@link EventHubClient} instance within the provided
+   * timeout. Use this method to perform clean-ups after the execution of the
+   * {@link EventHubClient}. Set timeout the {@link #BLOCK_UNTIL_CLOSE} to
+   * block until the client is closed.
+   *
+   * @param timeoutMs Close timeout in Milliseconds
+   */
+  void close(long timeoutMs);
+}

samza-azure/src/main/java/org/apache/samza/system/eventhub/EventHubClientManagerFactory.java

@@ -0,0 +1,32 @@
+/*
+* 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.system.eventhub;
+
+public class EventHubClientManagerFactory {
+  public EventHubClientManager getEventHubClientManager(String systemName, String streamName, EventHubConfig config) {
+
+    String eventHubNamespace = config.getStreamNamespace(systemName, streamName);
+    String entityPath = config.getStreamEntityPath(systemName, streamName);
+    String sasKeyName = config.getStreamSasKeyName(systemName, streamName);
+    String sasToken = config.getStreamSasToken(systemName, streamName);
+
+    return new SamzaEventHubClientManager(eventHubNamespace, entityPath, sasKeyName, sasToken);
+  }
+}