dawidwys commented on a change in pull request #18575: URL: https://github.com/apache/flink/pull/18575#discussion_r796836332
########## File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/MultipleFuturesAvailabilityHelper.java ########## @@ -0,0 +1,77 @@ +/* + * 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.streaming.runtime.io; + +import java.util.concurrent.CompletableFuture; + +import static org.apache.flink.util.concurrent.FutureUtils.assertNoException; + +/** + * This class is semi-thread safe. Only method {@link #notifyCompletion()} is allowed to be executed + * from an outside of the task thread. + * + * <p>It solves a problem of a potential memory leak as described in FLINK-25728. In short we have + * to ensure, that if there is one input (future) that rarely (or never) completes, that such future + * would not prevent previously returned combined futures (like {@link + * CompletableFuture#anyOf(CompletableFuture[])} from being garbage collected. Additionally, we + * don't want to accumulate more and more completion stages on such rarely completed future, so we + * are registering {@link CompletableFuture#thenRun(Runnable)} only if it has not already been done. + * + * <p>Note {@link #resetToUnAvailable()} doesn't de register previously registered futures. If + * future was registered in the past, but for whatever reason now it is not, such future can still + * complete the newly created future. + * + * <p>It might be no longer needed after upgrading to JDK9 + * (https://bugs.openjdk.java.net/browse/JDK-8160402). + */ +public class MultipleFuturesAvailabilityHelper { Review comment: `@Internal` ########## File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorTestBase.java ########## @@ -0,0 +1,129 @@ +/* +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.streaming.api.operators; + +import org.apache.flink.api.common.ExecutionConfig; +import org.apache.flink.api.common.state.OperatorStateStore; +import org.apache.flink.api.connector.source.mocks.MockSourceReader; +import org.apache.flink.api.connector.source.mocks.MockSourceSplit; +import org.apache.flink.api.connector.source.mocks.MockSourceSplitSerializer; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.core.fs.CloseableRegistry; +import org.apache.flink.core.io.SimpleVersionedSerialization; +import org.apache.flink.runtime.execution.Environment; +import org.apache.flink.runtime.operators.coordination.MockOperatorEventGateway; +import org.apache.flink.runtime.operators.testutils.MockEnvironment; +import org.apache.flink.runtime.operators.testutils.MockEnvironmentBuilder; +import org.apache.flink.runtime.operators.testutils.MockInputSplitProvider; +import org.apache.flink.runtime.state.AbstractStateBackend; +import org.apache.flink.runtime.state.StateInitializationContext; +import org.apache.flink.runtime.state.StateInitializationContextImpl; +import org.apache.flink.runtime.state.TestTaskStateManager; +import org.apache.flink.runtime.state.memory.MemoryStateBackend; +import org.apache.flink.streaming.api.operators.source.TestingSourceOperator; +import org.apache.flink.streaming.runtime.tasks.SourceOperatorStreamTask; +import org.apache.flink.streaming.runtime.tasks.StreamMockEnvironment; +import org.apache.flink.streaming.util.MockOutput; +import org.apache.flink.streaming.util.MockStreamConfig; + +import org.junit.After; +import org.junit.Before; + +import java.util.ArrayList; +import java.util.Collections; + +import static org.junit.Assert.assertTrue; + +/** Base class for testing {@link SourceOperator}. */ +@SuppressWarnings("serial") +public class SourceOperatorTestBase { Review comment: `abstract`? ########## File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/MultipleFuturesAvailabilityHelper.java ########## @@ -0,0 +1,77 @@ +/* + * 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.streaming.runtime.io; + +import java.util.concurrent.CompletableFuture; + +import static org.apache.flink.util.concurrent.FutureUtils.assertNoException; + +/** + * This class is semi-thread safe. Only method {@link #notifyCompletion()} is allowed to be executed + * from an outside of the task thread. + * + * <p>It solves a problem of a potential memory leak as described in FLINK-25728. In short we have + * to ensure, that if there is one input (future) that rarely (or never) completes, that such future + * would not prevent previously returned combined futures (like {@link + * CompletableFuture#anyOf(CompletableFuture[])} from being garbage collected. Additionally, we + * don't want to accumulate more and more completion stages on such rarely completed future, so we + * are registering {@link CompletableFuture#thenRun(Runnable)} only if it has not already been done. + * + * <p>Note {@link #resetToUnAvailable()} doesn't de register previously registered futures. If + * future was registered in the past, but for whatever reason now it is not, such future can still + * complete the newly created future. + * + * <p>It might be no longer needed after upgrading to JDK9 + * (https://bugs.openjdk.java.net/browse/JDK-8160402). + */ +public class MultipleFuturesAvailabilityHelper { + private final CompletableFuture<?>[] futuresToCombine; + + private volatile CompletableFuture<?> availableFuture = new CompletableFuture<>(); + + public MultipleFuturesAvailabilityHelper(int size) { + futuresToCombine = new CompletableFuture[size]; + } + + /** @return combined future using anyOf logic */ + public CompletableFuture<?> getAvailableFuture() { + return availableFuture; + } + + public void resetToUnAvailable() { + if (availableFuture.isDone()) { + availableFuture = new CompletableFuture<>(); + } + } + + private void notifyCompletion() { + availableFuture.complete(null); + } + + /** + * Combine {@code availabilityFuture} using anyOf logic with other previously registered + * futures. + */ + public void anyOf(final int idx, CompletableFuture<?> availabilityFuture) { Review comment: nit: Would `or` make more sense instead of `anyOf`? In `anyOf` I'd expect a collection of futures. This is a really nitty nit though. ########## File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorTestBase.java ########## @@ -0,0 +1,129 @@ +/* +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.streaming.api.operators; + +import org.apache.flink.api.common.ExecutionConfig; +import org.apache.flink.api.common.state.OperatorStateStore; +import org.apache.flink.api.connector.source.mocks.MockSourceReader; +import org.apache.flink.api.connector.source.mocks.MockSourceSplit; +import org.apache.flink.api.connector.source.mocks.MockSourceSplitSerializer; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.core.fs.CloseableRegistry; +import org.apache.flink.core.io.SimpleVersionedSerialization; +import org.apache.flink.runtime.execution.Environment; +import org.apache.flink.runtime.operators.coordination.MockOperatorEventGateway; +import org.apache.flink.runtime.operators.testutils.MockEnvironment; +import org.apache.flink.runtime.operators.testutils.MockEnvironmentBuilder; +import org.apache.flink.runtime.operators.testutils.MockInputSplitProvider; +import org.apache.flink.runtime.state.AbstractStateBackend; +import org.apache.flink.runtime.state.StateInitializationContext; +import org.apache.flink.runtime.state.StateInitializationContextImpl; +import org.apache.flink.runtime.state.TestTaskStateManager; +import org.apache.flink.runtime.state.memory.MemoryStateBackend; +import org.apache.flink.streaming.api.operators.source.TestingSourceOperator; +import org.apache.flink.streaming.runtime.tasks.SourceOperatorStreamTask; +import org.apache.flink.streaming.runtime.tasks.StreamMockEnvironment; +import org.apache.flink.streaming.util.MockOutput; +import org.apache.flink.streaming.util.MockStreamConfig; + +import org.junit.After; +import org.junit.Before; + +import java.util.ArrayList; +import java.util.Collections; + +import static org.junit.Assert.assertTrue; + +/** Base class for testing {@link SourceOperator}. */ +@SuppressWarnings("serial") +public class SourceOperatorTestBase { Review comment: Or even better, could we convert it to e.g. `SourceOperatorTestContext` and apply composition instead of inheritance? There is no test logic in the class, but just a setup. We could probably make it extend from `BeforeEach/AfterEach` of Junit 5, but I'd be also fine with just calling the setup/close methods in classes that use this "base". -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
