Refactor and refinement.

Author: Fanoid

flink-ml-core/src/main/java/org/apache/flink/ml/common/sharedobjects/AbstractSharedObjectsOneInputStreamOperator.java

@@ -22,10 +22,10 @@
 
 import java.util.List;
 
-/** The base class for all {@link OneInputStreamOperator} if shared objects are required. */
+/** The base class for {@link OneInputStreamOperator}s where shared objects are accessed. */
 public abstract class AbstractSharedObjectsOneInputStreamOperator<IN, OUT>
         extends AbstractSharedObjectsStreamOperator<OUT>
         implements OneInputStreamOperator<IN, OUT> {
 
-    public abstract List<ItemReadRequest<?>> registerReadsInProcessElement();
+    public abstract List<ReadRequest<?>> readRequestsInProcessElement();
 }

flink-ml-core/src/main/java/org/apache/flink/ml/common/sharedobjects/AbstractSharedObjectsStreamOperator.java

@@ -23,7 +23,7 @@
 import java.util.UUID;
 
 /**
- * A base class of stream operators which support shared objects.
+ * A base class of stream operators where shared objects are required.
  *
  * <p>Official subclasses, i.e., {@link AbstractSharedObjectsOneInputStreamOperator} and {@link
  * AbstractSharedObjectsTwoInputStreamOperator}, are strongly recommended.
@@ -32,31 +32,25 @@
  */
 public abstract class AbstractSharedObjectsStreamOperator<OUT> extends AbstractStreamOperator<OUT> {
 
-    private final String sharedObjectsAccessorID;
-    private transient SharedObjectsContext sharedObjectsContext;
+    /**
+     * A unique identifier for the instance, which is kept unchanged between client side and
+     * runtime.
+     */
+    private final String accessorID;
 
-    protected AbstractSharedObjectsStreamOperator() {
-        super();
-        sharedObjectsAccessorID = getClass().getSimpleName() + "-" + UUID.randomUUID();
-    }
-
-    protected void onSharedObjectsContextSet(SharedObjectsContext context) {
-        sharedObjectsContext = context;
-    }
+    /** The context for shared objects reads/writes. */
+    protected transient SharedObjectsContext context;
 
-    public String getSharedObjectsAccessorID() {
-        return sharedObjectsAccessorID;
-    }
-
-    public <T> T getSharedItem(ItemReadRequest<T> request) {
-        return sharedObjectsContext.get(request);
+    AbstractSharedObjectsStreamOperator() {
+        super();
+        accessorID = getClass().getSimpleName() + "-" + UUID.randomUUID();
     }
 
-    public <T> void setSharedItem(ItemDescriptor<T> key, T value) {
-        sharedObjectsContext.set(key, value);
+    void onSharedObjectsContextSet(SharedObjectsContext context) {
+        this.context = context;
     }
 
-    public <T> void renewSharedItem(ItemDescriptor<T> key) {
-        sharedObjectsContext.renew(key);
+    String getAccessorID() {
+        return accessorID;
     }
 }

flink-ml-core/src/main/java/org/apache/flink/ml/common/sharedobjects/AbstractSharedObjectsTwoInputStreamOperator.java

@@ -22,12 +22,12 @@
 
 import java.util.List;
 
-/** The base class for all {@link TwoInputStreamOperator} if shared objects are required. */
+/** The base class for {@link TwoInputStreamOperator}s where shared objects are accessed. */
 public abstract class AbstractSharedObjectsTwoInputStreamOperator<IN1, IN2, OUT>
         extends AbstractSharedObjectsStreamOperator<OUT>
         implements TwoInputStreamOperator<IN1, IN2, OUT> {
 
-    public abstract List<ItemReadRequest<?>> registerReadsInProcessElement1();
+    public abstract List<ReadRequest<?>> readRequestsInProcessElement1();
 
-    public abstract List<ItemReadRequest<?>> registerReadsInProcessElement2();
+    public abstract List<ReadRequest<?>> readRequestsInProcessElement2();
 }

flink-ml-core/src/main/java/org/apache/flink/ml/common/sharedobjects/AbstractSharedObjectsWrapperOperator.java

@@ -91,7 +91,7 @@ abstract class AbstractSharedObjectsWrapperOperator<
     private final int numInputs;
     private final TypeSerializer<?>[] inTypeSerializers;
     private final ListStateWithCache<CacheElement<?>>[] cachedElements;
-    private final Queue<ItemReadRequest<?>>[] itemReadRequests;
+    private final Queue<ReadRequest<?>>[] readRequests;
     private final boolean[] hasCachedElements;
 
     protected transient StreamOperatorStateHandler stateHandler;
@@ -131,11 +131,11 @@ abstract class AbstractSharedObjectsWrapperOperator<
         numInputs = numNetworkInputs;
 
         inTypeSerializers = new TypeSerializer[numInputs];
-        itemReadRequests = new Queue[numInputs];
+        readRequests = new Queue[numInputs];
         for (int i = 0; i < numInputs; i++) {
             inTypeSerializers[i] =
                     streamConfig.getTypeSerializerIn(i, containingTask.getUserCodeClassLoader());
-            itemReadRequests[i] = new ArrayDeque<>(getInputItemReadRequests(i));
+            readRequests[i] = new ArrayDeque<>(getInputReadRequests(i));
         }
         cachedElements = new ListStateWithCache[numInputs];
         hasCachedElements = new boolean[numInputs];
@@ -162,18 +162,18 @@ private OperatorMetricGroup createOperatorMetricGroup(
     }
 
     /**
-     * Checks if the item read requests are satisfied for the input.
+     * Checks if the read requests are satisfied for the input.
      *
      * @param inputId The input id, starting from 0.
      * @param wait Whether to wait until all requests satisfied, or not.
      * @return If all requests of this input are satisfied.
      */
-    private boolean checkSharedItemReadsReady(int inputId, boolean wait) {
-        Queue<ItemReadRequest<?>> requests = itemReadRequests[inputId];
+    private boolean checkReadRequestsReady(int inputId, boolean wait) {
+        Queue<ReadRequest<?>> requests = readRequests[inputId];
         while (!requests.isEmpty()) {
-            ItemReadRequest<?> request = requests.poll();
+            ReadRequest<?> request = requests.poll();
             try {
-                if (null == context.get(request, wait)) {
+                if (null == context.read(request, wait)) {
                     requests.add(request);
                     return false;
                 }
@@ -185,12 +185,12 @@ private boolean checkSharedItemReadsReady(int inputId, boolean wait) {
     }
 
     /**
-     * Gets {@link ItemReadRequest}s required for processing elements in the input.
+     * Gets {@link ReadRequest}s required for processing elements in the input.
      *
      * @param inputId The input id, starting from 0.
-     * @return The {@link ItemReadRequest}s required for processing elements.
+     * @return The {@link ReadRequest}s required for processing elements.
      */
-    protected abstract List<ItemReadRequest<?>> getInputItemReadRequests(int inputId);
+    protected abstract List<ReadRequest<?>> getInputReadRequests(int inputId);
 
     /**
      * Extracts common processing logic in subclasses' processing elements.
@@ -213,7 +213,7 @@ protected void processElementX(
             ThrowingConsumer<Watermark, Exception> watermarkConsumer,
             ThrowingConsumer<StreamRecord, Exception> keyContextSetter)
             throws Exception {
-        if (checkSharedItemReadsReady(inputId, false)) {
+        if (checkReadRequestsReady(inputId, false)) {
             if (hasCachedElements[inputId]) {
                 processCachedElements(
                         inputId, elementConsumer, watermarkConsumer, keyContextSetter);
@@ -248,7 +248,7 @@ protected void processWatermarkX(
             ThrowingConsumer<Watermark, Exception> watermarkConsumer,
             ThrowingConsumer<StreamRecord, Exception> keyContextSetter)
             throws Exception {
-        if (checkSharedItemReadsReady(inputId, false)) {
+        if (checkReadRequestsReady(inputId, false)) {
             if (hasCachedElements[inputId]) {
                 processCachedElements(
                         inputId, elementConsumer, watermarkConsumer, keyContextSetter);
@@ -281,7 +281,7 @@ protected void endInputX(
             ThrowingConsumer<StreamRecord, Exception> keyContextSetter)
             throws Exception {
         if (hasCachedElements[inputId]) {
-            checkSharedItemReadsReady(inputId, true);
+            checkReadRequestsReady(inputId, true);
             processCachedElements(inputId, elementConsumer, watermarkConsumer, keyContextSetter);
             hasCachedElements[inputId] = false;
         }
@@ -322,8 +322,8 @@ private void processCachedElements(
             }
         }
         cachedElements[inputId].clear();
-        Preconditions.checkState(itemReadRequests[inputId].isEmpty());
-        itemReadRequests[inputId].addAll(getInputItemReadRequests(inputId));
+        Preconditions.checkState(readRequests[inputId].isEmpty());
+        readRequests[inputId].addAll(getInputReadRequests(inputId));
     }
 
     @Override
@@ -489,7 +489,7 @@ public void onEpochWatermarkIncremented(
         for (int i = 0; i < numInputs; i += 1) {
             processCachedElementsBeforeEpochIncremented(i);
         }
-        this.context.increaseEpoch(epochWatermark);
+        this.context.incStep(epochWatermark);
         if (wrappedOperator instanceof IterationListener) {
             //noinspection unchecked
             ((IterationListener<T>) wrappedOperator)
@@ -499,7 +499,7 @@ public void onEpochWatermarkIncremented(
 
     @Override
     public void onIterationTerminated(Context context, Collector<T> collector) throws Exception {
-        this.context.increaseEpoch();
+        this.context.incStep();
         if (wrappedOperator instanceof IterationListener) {
             //noinspection unchecked
             ((IterationListener<T>) wrappedOperator).onIterationTerminated(context, collector);

flink-ml-core/src/main/java/org/apache/flink/ml/common/sharedobjects/ItemDescriptor.java renamed to flink-ml-core/src/main/java/org/apache/flink/ml/common/sharedobjects/Descriptor.java

@@ -27,14 +27,17 @@
 import java.io.Serializable;
 
 /**
- * Descriptor for a shared item.
+ * Descriptor for a shared object.
  *
- * @param <T> The type of the shared item.
+ * <p>A shared object can have a non-null initial value, or have no initial values. If a non-null
+ * initial value provided, it is set with an initial write-step (See {@link ReadRequest}).
+ *
+ * @param <T> The type of the shared object.
  */
 @Experimental
-public class ItemDescriptor<T> implements Serializable {
+public class Descriptor<T> implements Serializable {
 
-    /** Name of the item. */
+    /** Name of the shared object. */
     public final String name;
 
     /** Type serializer. */
@@ -43,32 +46,50 @@ public class ItemDescriptor<T> implements Serializable {
     /** Initialize value. */
     public final @Nullable T initVal;
 
-    private ItemDescriptor(String name, TypeSerializer<T> serializer, T initVal) {
+    private Descriptor(String name, TypeSerializer<T> serializer, T initVal) {
         this.name = name;
         this.serializer = serializer;
         this.initVal = initVal;
     }
 
-    public static <T> ItemDescriptor<T> of(String name, TypeSerializer<T> serializer, T initVal) {
+    public static <T> Descriptor<T> of(String name, TypeSerializer<T> serializer, T initVal) {
         Preconditions.checkNotNull(
-                initVal, "Cannot use `null` as the initial value of a shared item.");
-        return new ItemDescriptor<>(name, serializer, initVal);
+                initVal, "Cannot use `null` as the initial value of a shared object.");
+        return new Descriptor<>(name, serializer, initVal);
     }
 
-    public static <T> ItemDescriptor<T> of(String name, TypeSerializer<T> serializer) {
-        return new ItemDescriptor<>(name, serializer, null);
+    public static <T> Descriptor<T> of(String name, TypeSerializer<T> serializer) {
+        return new Descriptor<>(name, serializer, null);
     }
 
-    public ItemReadRequest<T> sameEpoch() {
-        return new ItemReadRequest<>(this, ItemReadRequest.OFFSET.SAME);
+    /**
+     * Creates a read request which always reads this shared object with same read-step as the
+     * operator step.
+     *
+     * @return A read request.
+     */
+    public ReadRequest<T> sameStep() {
+        return new ReadRequest<>(this, ReadRequest.OFFSET.SAME);
     }
 
-    public ItemReadRequest<T> prevEpoch() {
-        return new ItemReadRequest<>(this, ItemReadRequest.OFFSET.PREV);
+    /**
+     * Creates a read request which always reads this shared object with the read-step be the
+     * previous item of the operator step.
+     *
+     * @return A read request.
+     */
+    public ReadRequest<T> prevStep() {
+        return new ReadRequest<>(this, ReadRequest.OFFSET.PREV);
     }
 
-    public ItemReadRequest<T> nextEpoch() {
-        return new ItemReadRequest<>(this, ItemReadRequest.OFFSET.NEXT);
+    /**
+     * Creates a read request which always reads this shared object with the read-step be the next
+     * item of the operator step.
+     *
+     * @return A read request.
+     */
+    public ReadRequest<T> nextStep() {
+        return new ReadRequest<>(this, ReadRequest.OFFSET.NEXT);
     }
 
     @Override
@@ -84,13 +105,13 @@ public boolean equals(Object o) {
         if (o == null || getClass() != o.getClass()) {
             return false;
         }
-        ItemDescriptor<?> that = (ItemDescriptor<?>) o;
+        Descriptor<?> that = (Descriptor<?>) o;
         return name.equals(that.name);
     }
 
     @Override
     public String toString() {
         return String.format(
-                "ItemDescriptor{name='%s', serializer=%s, initVal=%s}", name, serializer, initVal);
+                "Descriptor{name='%s', serializer=%s, initVal=%s}", name, serializer, initVal);
     }
 }

flink-ml-core/src/main/java/org/apache/flink/ml/common/sharedobjects/ItemReadRequest.java

@@ -45,9 +45,9 @@ class OneInputSharedObjectsWrapperOperator<IN, OUT>
     }
 
     @Override
-    protected List<ItemReadRequest<?>> getInputItemReadRequests(int inputId) {
+    protected List<ReadRequest<?>> getInputReadRequests(int inputId) {
         Preconditions.checkArgument(0 == inputId);
-        return wrappedOperator.registerReadsInProcessElement();
+        return wrappedOperator.readRequestsInProcessElement();
     }
 
     @Override