Resolving deadlock.

Author: Fanoid

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

@@ -18,33 +18,14 @@
 
 package org.apache.flink.ml.common.sharedobjects;
 
-import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
 import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
-import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
 
-import java.lang.reflect.Method;
-import java.util.Collections;
 import java.util.List;
-import java.util.Map;
 
-/**
- * A default implementation of {@link AbstractStreamOperator} which implements {@link
- * SharedObjectsStreamOperator}. Use this class to reduce boilerplate codes.
- */
+/** The base class for all {@link OneInputStreamOperator} if shared objects are required. */
 public abstract class AbstractSharedObjectsOneInputStreamOperator<IN, OUT>
         extends AbstractSharedObjectsStreamOperator<OUT>
         implements OneInputStreamOperator<IN, OUT> {
 
     public abstract List<ItemReadRequest<?>> registerReadsInProcessElement();
-
-    @Override
-    public Map<Method, List<ItemReadRequest<?>>> registerReads() {
-        try {
-            return Collections.singletonMap(
-                    OneInputStreamOperator.class.getMethod("processElement", StreamRecord.class),
-                    registerReadsInProcessElement());
-        } catch (NoSuchMethodException e) {
-            throw new RuntimeException(e);
-        }
-    }
 }

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

@@ -23,11 +23,14 @@
 import java.util.UUID;
 
 /**
- * A default implementation of {@link AbstractStreamOperator} which implements {@link
- * SharedObjectsStreamOperator}.
+ * A base class of stream operators which support shared objects.
+ *
+ * <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.
  */
-abstract class AbstractSharedObjectsStreamOperator<OUT> extends AbstractStreamOperator<OUT>
-        implements SharedObjectsStreamOperator {
+public abstract class AbstractSharedObjectsStreamOperator<OUT> extends AbstractStreamOperator<OUT> {
 
     private final String sharedObjectsAccessorID;
     private transient SharedObjectsContext sharedObjectsContext;
@@ -37,7 +40,7 @@ protected AbstractSharedObjectsStreamOperator() {
         sharedObjectsAccessorID = getClass().getSimpleName() + "-" + UUID.randomUUID();
     }
 
-    public void onSharedObjectsContextSet(SharedObjectsContext context) {
+    protected void onSharedObjectsContextSet(SharedObjectsContext context) {
         sharedObjectsContext = context;
     }
 

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

@@ -18,40 +18,16 @@
 
 package org.apache.flink.ml.common.sharedobjects;
 
-import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
 import org.apache.flink.streaming.api.operators.TwoInputStreamOperator;
-import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
 
-import java.lang.reflect.Method;
-import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
 
-/**
- * A default implementation of {@link AbstractStreamOperator} which implements {@link
- * SharedObjectsStreamOperator}. Use this class to reduce boilerplate codes.
- */
+/** The base class for all {@link TwoInputStreamOperator} if shared objects are required. */
 public abstract class AbstractSharedObjectsTwoInputStreamOperator<IN1, IN2, OUT>
         extends AbstractSharedObjectsStreamOperator<OUT>
         implements TwoInputStreamOperator<IN1, IN2, OUT> {
 
     public abstract List<ItemReadRequest<?>> registerReadsInProcessElement1();
 
     public abstract List<ItemReadRequest<?>> registerReadsInProcessElement2();
-
-    @Override
-    public Map<Method, List<ItemReadRequest<?>>> registerReads() {
-        Map<Method, List<ItemReadRequest<?>>> m = new HashMap<>();
-        try {
-            m.put(
-                    TwoInputStreamOperator.class.getMethod("processElement1", StreamRecord.class),
-                    registerReadsInProcessElement1());
-            m.put(
-                    TwoInputStreamOperator.class.getMethod("processElement2", StreamRecord.class),
-                    registerReadsInProcessElement2());
-        } catch (NoSuchMethodException e) {
-            throw new RuntimeException(e);
-        }
-        return m;
-    }
 }