Refactor share objects infra to resolve challenges.

Author: Fanoid

…link/ml/common/sharedobjects/PoolID.java …SharedObjectsOneInputStreamOperator.javaflink-ml-core/src/main/java/org/apache/flink/ml/common/sharedobjects/PoolID.java renamed to flink-ml-core/src/main/java/org/apache/flink/ml/common/sharedobjects/AbstractSharedObjectsOneInputStreamOperator.java flink-ml-core/src/main/java/org/apache/flink/ml/common/sharedobjects/PoolID.java renamed to flink-ml-core/src/main/java/org/apache/flink/ml/common/sharedobjects/AbstractSharedObjectsOneInputStreamOperator.java

@@ -18,15 +18,14 @@
 
 package org.apache.flink.ml.common.sharedobjects;
 
-import org.apache.flink.util.AbstractID;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
 
-/** ID of a pool for shared objects. */
-class PoolID extends AbstractID {
-    private static final long serialVersionUID = 1L;
+import java.util.List;
 
-    public PoolID(byte[] bytes) {
-        super(bytes);
-    }
+/** 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 PoolID() {}
+    public abstract List<ReadRequest<?>> readRequestsInProcessElement();
 }

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

@@ -0,0 +1,56 @@
+/*
+ * 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.flink.ml.common.sharedobjects;
+
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+
+import java.util.UUID;
+
+/**
+ * A base class of stream operators where shared objects are required.
+ *
+ * <p>Official subclasses, i.e., {@link AbstractSharedObjectsOneInputStreamOperator} and {@link
+ * AbstractSharedObjectsTwoInputStreamOperator}, are strongly recommended.
+ *
+ * <p>If you are going to implement a subclass by yourself, you have to handle potential deadlocks.
+ */
+public abstract class AbstractSharedObjectsStreamOperator<OUT> extends AbstractStreamOperator<OUT> {
+
+    /**
+     * A unique identifier for the instance, which is kept unchanged between client side and
+     * runtime.
+     */
+    private final String accessorID;
+
+    /** The context for shared objects reads/writes. */
+    protected transient SharedObjectsContext context;
+
+    AbstractSharedObjectsStreamOperator() {
+        super();
+        accessorID = getClass().getSimpleName() + "-" + UUID.randomUUID();
+    }
+
+    void onSharedObjectsContextSet(SharedObjectsContext context) {
+        this.context = context;
+    }
+
+    String getAccessorID() {
+        return accessorID;
+    }
+}

…objects/SharedObjectsStreamOperator.java …SharedObjectsTwoInputStreamOperator.javaflink-ml-core/src/main/java/org/apache/flink/ml/common/sharedobjects/SharedObjectsStreamOperator.java renamed to flink-ml-core/src/main/java/org/apache/flink/ml/common/sharedobjects/AbstractSharedObjectsTwoInputStreamOperator.java flink-ml-core/src/main/java/org/apache/flink/ml/common/sharedobjects/SharedObjectsStreamOperator.java renamed to flink-ml-core/src/main/java/org/apache/flink/ml/common/sharedobjects/AbstractSharedObjectsTwoInputStreamOperator.java

@@ -18,21 +18,16 @@
 
 package org.apache.flink.ml.common.sharedobjects;
 
-/** Interface for all operators that need to access the shared objects. */
-public interface SharedObjectsStreamOperator {
+import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
 
-    /**
-     * Set the shared objects context in runtime.
-     *
-     * @param context The context for shared objects.
-     */
-    void onSharedObjectsContextSet(SharedObjectsContext context);
+import java.util.List;
 
-    /**
-     * Get a unique ID to represent the operator instance. The ID must be kept unchanged through its
-     * lifetime.
-     *
-     * @return A unique ID.
-     */
-    String getSharedObjectsAccessorID();
+/** 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<ReadRequest<?>> readRequestsInProcessElement1();
+
+    public abstract List<ReadRequest<?>> readRequestsInProcessElement2();
 }