Merge remote-tracking branch 'upstream/master'

Author: weisong44

samza-api/src/main/java/org/apache/samza/operators/TimerRegistry.java renamed to samza-api/src/main/java/org/apache/samza/operators/Scheduler.java

@@ -20,22 +20,20 @@
 package org.apache.samza.operators;
 
 /**
- * Allows registering epoch-time timer callbacks from the operators.
- * See {@link org.apache.samza.operators.functions.TimerFunction} for details.
- * @param <K> type of the timer key
+ * Allows scheduling {@link org.apache.samza.operators.functions.ScheduledFunction} callbacks to be invoked later.
+ * @param <K> type of the key to schedule
  */
-public interface TimerRegistry<K> {
-
+public interface Scheduler<K> {
   /**
-   * Register a epoch-time timer with key.
-   * @param key unique timer key
-   * @param timestamp epoch time when the timer will be fired, in milliseconds
+   * Schedule a callback for the {@code key} to be invoked at {@code timestamp}.
+   * @param key unique key associated with the callback to schedule
+   * @param timestamp epoch time when the callback for the key will be invoked, in milliseconds
    */
-  void register(K key, long timestamp);
+  void schedule(K key, long timestamp);
 
   /**
-   * Delete the timer for the provided key.
-   * @param key key for the timer to delete
+   * Delete the scheduled callback for the provided {@code key}.
+   * @param key key to delete
    */
   void delete(K key);
 }

samza-api/src/main/java/org/apache/samza/operators/functions/ScheduledFunction.java

@@ -0,0 +1,73 @@
+/*
+ * 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.operators.functions;
+
+import org.apache.samza.operators.Scheduler;
+
+import java.util.Collection;
+
+
+/**
+ * Allows scheduling a callback for a specific epoch-time.
+ * Key must be a unique identifier for its corresponding logic to execute, and is provided in the callback when the
+ * corresponding schedule time occurs.
+ *
+ * <p>
+ * Example of a {@link FlatMapFunction} with {@link ScheduledFunction}:
+ * <pre>{@code
+ *    public class ExampleScheduledFn implements FlatMapFunction<String, String>, ScheduledFunction<String, String> {
+ *      // for recurring callbacks, keep track of the scheduler from "schedule"
+ *      private Scheduler scheduler;
+ *
+ *      public void schedule(Scheduler scheduler) {
+ *        // save the scheduler for recurring callbacks
+ *        this.scheduler = scheduler;
+ *        long time = System.currentTimeMillis() + 5000; // fire after 5 sec
+ *        scheduler.schedule("do-delayed-logic", time);
+ *      }
+ *      public Collection<String> apply(String s) {
+ *        ...
+ *      }
+ *      public Collection<String> onCallback(String key, long timestamp) {
+ *        // do some logic for key "do-delayed-logic"
+ *        ...
+ *        // for recurring callbacks, call the saved scheduler again
+ *        this.scheduler.schedule("example-process", System.currentTimeMillis() + 5000);
+ *      }
+ *    }
+ * }</pre>
+ * @param <K> type of the key
+ * @param <OM> type of the output
+ */
+public interface ScheduledFunction<K, OM> {
+  /**
+   * Allows scheduling the initial callback(s) and saving the {@code scheduler} for later use for recurring callbacks.
+   * @param scheduler used to specify the schedule time(s) and key(s)
+   */
+  void schedule(Scheduler<K> scheduler);
+
+  /**
+   * Returns the output from the scheduling logic corresponding to the key that was triggered.
+   * @param key key corresponding to the callback that got invoked
+   * @param timestamp schedule time that was set for the callback for the key, in milliseconds since epoch
+   * @return {@link Collection} of output elements
+   */
+  Collection<OM> onCallback(K key, long timestamp);
+}

samza-api/src/main/java/org/apache/samza/operators/functions/TimerFunction.java

@@ -17,18 +17,23 @@
  * under the License.
  */
 
-package org.apache.samza.task;
+package org.apache.samza.scheduler;
+
+import org.apache.samza.task.MessageCollector;
+import org.apache.samza.task.TaskCoordinator;
+
 
 /**
- * The callback that is invoked when its corresponding timer registered via {@link TaskContext} fires.
- * @param <K> type of the timer key
+ * The callback that is invoked when its corresponding schedule time registered via
+ * {@link org.apache.samza.task.TaskContext} is reached.
+ * @param <K> type of the callback key
  */
-public interface TimerCallback<K> {
+public interface ScheduledCallback<K> {
   /**
-   * Invoked when the timer of key fires.
-   * @param key timer key
+   * Invoked when the corresponding schedule time is reached.
+   * @param key key for callback
    * @param collector contains the means of sending message envelopes to the output stream.
    * @param coordinator manages execution of tasks.
    */
-  void onTimer(K key, MessageCollector collector, TaskCoordinator coordinator);
+  void onCallback(K key, MessageCollector collector, TaskCoordinator coordinator);
 }

samza-api/src/main/java/org/apache/samza/task/TimerCallback.java renamed to samza-api/src/main/java/org/apache/samza/scheduler/ScheduledCallback.java

@@ -24,6 +24,7 @@
 import org.apache.samza.container.SamzaContainerContext;
 import org.apache.samza.container.TaskName;
 import org.apache.samza.metrics.MetricsRegistry;
+import org.apache.samza.scheduler.ScheduledCallback;
 import org.apache.samza.system.SystemStreamPartition;
 import org.apache.samza.table.Table;
 
@@ -76,21 +77,21 @@ default Object getUserContext() {
   }
 
   /**
-   * Register a keyed timer with a callback of {@link TimerCallback} in this task.
+   * Schedule the {@code callback} for the provided {@code key} to be invoked at epoch-time {@code timestamp}.
    * The callback will be invoked exclusively with any other operations for this task,
    * e.g. processing, windowing and commit.
-   * @param key timer key
-   * @param timestamp epoch time when the timer will be fired, in milliseconds
-   * @param callback callback when the timer is fired
+   * @param key key for the callback
+   * @param timestamp epoch time when the callback will be fired, in milliseconds
+   * @param callback callback to call when the {@code timestamp} is reached
    * @param <K> type of the key
    */
-  <K> void registerTimer(K key, long timestamp, TimerCallback<K> callback);
+  <K> void scheduleCallback(K key, long timestamp, ScheduledCallback<K> callback);
 
   /**
-   * Delete the keyed timer in this task.
-   * Deletion only happens if the timer hasn't been fired. Otherwise it will not interrupt.
-   * @param key timer key
+   * Delete the scheduled {@code callback} for the {@code key}.
+   * Deletion only happens if the callback hasn't been fired. Otherwise it will not interrupt.
+   * @param key callback key
    * @param <K> type of the key
    */
-  <K> void deleteTimer(K key);
+  <K> void deleteScheduledCallback(K key);
 }

samza-api/src/main/java/org/apache/samza/task/TaskContext.java

@@ -65,19 +65,12 @@ public EventHubSystemAdmin(String systemName, EventHubConfig eventHubConfig,
     this.eventHubClientManagerFactory = eventHubClientManagerFactory;
   }
 
-  private String getNextOffset(String currentOffset) {
-    // EventHub will return the first message AFTER the offset
-    // that was specified in the fetch request.
-    // If no such offset exists Eventhub will return an error.
-    return String.valueOf(Long.parseLong(currentOffset) + 1);
-  }
-
   @Override
   public Map<SystemStreamPartition, String> getOffsetsAfter(Map<SystemStreamPartition, String> offsets) {
-    Map<SystemStreamPartition, String> results = new HashMap<>();
-
-    offsets.forEach((partition, offset) -> results.put(partition, getNextOffset(offset)));
-    return results;
+    // In EventHubSystemConsumer#initializeEventHubsManagers, we exclude the offset that we specify. i.e.
+    // we will only get the message after the checkpoint offset. Hence, by returning the same offset as the
+    // "next" offset, we won't be reprocessing the same event.
+    return offsets;
   }
 
   // EventHubRuntimeInformation does not implement toString()

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

@@ -276,13 +276,11 @@ private synchronized void initializeEventHubsManagers() {
           receiver = eventHubClientManager.getEventHubClient()
               .createReceiverSync(consumerGroup, partitionId.toString(), EventPosition.fromEnqueuedTime(Instant.now()));
         } else {
-          // If the offset is less or equal to the newest offset in the system, it can be
-          // used as the starting offset to receive from. EventHub will return the first
-          // message AFTER the offset that was specified in the fetch request.
+          // EventHub will return the first message AFTER the offset that was specified in the fetch request.
           // If no such offset exists Eventhub will return an error.
           receiver = eventHubClientManager.getEventHubClient()
               .createReceiverSync(consumerGroup, partitionId.toString(),
-                  EventPosition.fromOffset(offset, !offset.equals(EventHubSystemConsumer.START_OF_STREAM)));
+                  EventPosition.fromOffset(offset, /* inclusiveFlag */false));
         }
 
         receiver.setPrefetchCount(prefetchCount);

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

@@ -22,7 +22,6 @@
 import org.apache.samza.Partition;
 import org.apache.samza.system.SystemAdmin;
 import org.apache.samza.system.SystemStreamMetadata;
-import org.apache.samza.system.SystemStreamPartition;
 import org.apache.samza.system.eventhub.EventHubSystemFactory;
 import org.apache.samza.system.eventhub.MockEventHubConfigFactory;
 import org.apache.samza.system.eventhub.consumer.EventHubSystemConsumer;
@@ -31,7 +30,6 @@
 import org.junit.Ignore;
 import org.junit.Test;
 
-import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
@@ -53,23 +51,6 @@ public void testOffsetComparison() {
     Assert.assertNull(eventHubSystemAdmin.offsetComparator(EventHubSystemConsumer.END_OF_STREAM, EventHubSystemConsumer.END_OF_STREAM));
   }
 
-  @Test
-  public void testGetNextOffset() {
-    EventHubSystemFactory eventHubSystemFactory = new EventHubSystemFactory();
-    SystemAdmin eventHubSystemAdmin = eventHubSystemFactory.getAdmin(SYSTEM_NAME,
-            MockEventHubConfigFactory.getEventHubConfig(EventHubSystemProducer.PartitioningMethod.EVENT_HUB_HASHING));
-    Map<SystemStreamPartition, String> offsets = new HashMap<>();
-    SystemStreamPartition ssp0 = new SystemStreamPartition(SYSTEM_NAME, STREAM_NAME1, new Partition(0));
-    SystemStreamPartition ssp2 = new SystemStreamPartition(SYSTEM_NAME, STREAM_NAME1, new Partition(2));
-    offsets.put(ssp0, Integer.toString(0));
-    offsets.put(ssp2, EventHubSystemConsumer.START_OF_STREAM);
-
-    Map<SystemStreamPartition, String> updatedOffsets = eventHubSystemAdmin.getOffsetsAfter(offsets);
-    Assert.assertEquals(offsets.size(), updatedOffsets.size());
-    Assert.assertEquals("1", updatedOffsets.get(ssp0));
-    Assert.assertEquals("0", updatedOffsets.get(ssp2));
-  }
-
   @Ignore("Integration Test")
   @Test
   public void testGetStreamMetadata() {

samza-azure/src/test/java/org/apache/samza/system/eventhub/admin/TestEventHubSystemAdmin.java

@@ -29,9 +29,9 @@
 import org.apache.samza.system.SystemStreamPartition;
 import org.apache.samza.table.Table;
 import org.apache.samza.table.TableManager;
-import org.apache.samza.task.SystemTimerScheduler;
+import org.apache.samza.task.EpochTimeScheduler;
 import org.apache.samza.task.TaskContext;
-import org.apache.samza.task.TimerCallback;
+import org.apache.samza.scheduler.ScheduledCallback;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -53,7 +53,7 @@ public class TaskContextImpl implements TaskContext {
   private final JobModel jobModel;
   private final StreamMetadataCache streamMetadataCache;
   private final Map<String, Object> objectRegistry = new HashMap<>();
-  private final SystemTimerScheduler timerScheduler;
+  private final EpochTimeScheduler timerScheduler;
 
   private Object userContext = null;
 
@@ -76,7 +76,7 @@ public TaskContextImpl(TaskName taskName,
     this.tableManager = tableManager;
     this.jobModel = jobModel;
     this.streamMetadataCache = streamMetadataCache;
-    this.timerScheduler = SystemTimerScheduler.create(timerExecutor);
+    this.timerScheduler = EpochTimeScheduler.create(timerExecutor);
   }
 
   @Override
@@ -134,12 +134,12 @@ public Object getUserContext() {
   }
 
   @Override
-  public <K> void registerTimer(K key, long timestamp, TimerCallback<K> callback) {
+  public <K> void scheduleCallback(K key, long timestamp, ScheduledCallback<K> callback) {
     timerScheduler.setTimer(key, timestamp, callback);
   }
 
   @Override
-  public <K> void deleteTimer(K key) {
+  public <K> void deleteScheduledCallback(K key) {
     timerScheduler.deleteTimer(key);
   }
 
@@ -159,7 +159,7 @@ public StreamMetadataCache getStreamMetadataCache() {
     return streamMetadataCache;
   }
 
-  public SystemTimerScheduler getTimerScheduler() {
+  public EpochTimeScheduler getTimerScheduler() {
     return timerScheduler;
   }
 }