This is an automated email from the ASF dual-hosted git repository.
rmetzger pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 4a0abed7151 [FLINK-39330][runtime] Make CollectSinkFunction teardown
null-safe (#28474)
4a0abed7151 is described below
commit 4a0abed715192d6f6c18708aa9b15f41d525ef40
Author: Abhijeet Sharma <[email protected]>
AuthorDate: Fri Oct 30 19:08:11 2026 +0530
[FLINK-39330][runtime] Make CollectSinkFunction teardown null-safe (#28474)
When a CollectSinkOperator is torn down before initializeState/open ran
(e.g. during a JobManager failover), CollectSinkOperator#close() invokes
accumulateFinalResults() on a sink function whose bufferLock (and
serverThread) were never initialized, throwing a NullPointerException.
That NPE surfaces in the task's exception-handler path and is escalated
to a fatal error that shuts down the TaskManager. In a single-TM
MiniCluster the recovered job can then never reacquire slots, starves for
the full slot request timeout, and fails with NoResourceAvailableException.
Guard accumulateFinalResults() and close() against being called before the
function was opened so the failover teardown no longer raises an NPE.
Add tests for closing a function that was never opened, and for closing
one whose state was initialized but which was never opened; the latter
also pins down the resulting accumulator content (offset 0, null version,
no results).
Generated-by: Claude Code (Claude Opus 5.5)
Co-authored-by: Abhijeet Sharma <[email protected]>
---
.../api/operators/collect/CollectSinkFunction.java | 9 +++++--
.../operators/collect/CollectSinkFunctionTest.java | 28 ++++++++++++++++++++++
.../utils/CollectSinkFunctionTestWrapper.java | 9 +++++--
3 files changed, 42 insertions(+), 4 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/collect/CollectSinkFunction.java
b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/collect/CollectSinkFunction.java
index 1062cc21aaa..fdb3a2ea1db 100644
---
a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/collect/CollectSinkFunction.java
+++
b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/collect/CollectSinkFunction.java
@@ -300,11 +300,16 @@ public class CollectSinkFunction<IN> extends
RichSinkFunction<IN>
@Override
public void close() throws Exception {
- serverThread.close();
- serverThread.join();
+ if (serverThread != null) {
+ serverThread.close();
+ serverThread.join();
+ }
}
public void accumulateFinalResults() throws Exception {
+ if (bufferLock == null) {
+ return;
+ }
bufferLock.lock();
try {
// put results not consumed by the client into the accumulator
diff --git
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/collect/CollectSinkFunctionTest.java
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/collect/CollectSinkFunctionTest.java
index 34efd510a6d..fcc1338627c 100644
---
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/collect/CollectSinkFunctionTest.java
+++
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/collect/CollectSinkFunctionTest.java
@@ -34,6 +34,7 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.List;
+import static org.assertj.core.api.Assertions.assertThatCode;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link CollectSinkFunction}. */
@@ -54,6 +55,33 @@ class CollectSinkFunctionTest {
functionWrapper.closeWrapper();
}
+ @Test
+ void testCloseBeforeOpenDoesNotThrow() {
+ CollectSinkFunction<Integer> function =
+ new CollectSinkFunction<>(
+ serializer, 12,
CollectSinkFunctionTestWrapper.ACCUMULATOR_NAME);
+ assertThatCode(
+ () -> {
+ function.accumulateFinalResults();
+ function.close();
+ })
+ .doesNotThrowAnyException();
+ }
+
+ @Test
+ void testCloseAfterInitializeStateBeforeOpenDoesNotThrow() throws
Exception {
+ functionWrapper.initializeFunctionWithState();
+
assertThatCode(functionWrapper::closeFunctionNormally).doesNotThrowAnyException();
+ // never opened, so no version was generated and nothing was buffered
+ CollectTestUtils.assertAccumulatorResult(
+ functionWrapper.getAccumulatorResults(),
+ 0,
+ null,
+ 0,
+ Collections.emptyList(),
+ serializer);
+ }
+
@Test
void testIncreasingToken() throws Exception {
functionWrapper.openFunction();
diff --git
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/collect/utils/CollectSinkFunctionTestWrapper.java
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/collect/utils/CollectSinkFunctionTestWrapper.java
index 2a7be271363..a24b462c84f 100644
---
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/collect/utils/CollectSinkFunctionTestWrapper.java
+++
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/collect/utils/CollectSinkFunctionTestWrapper.java
@@ -105,13 +105,18 @@ public class CollectSinkFunctionTestWrapper<IN> {
}
public void openFunctionWithState() throws Exception {
+ initializeFunctionWithState();
+ function.open(DefaultOpenContext.INSTANCE);
+ coordinator.handleEventFromOperator(0, 0, gateway.getNextEvent());
+ }
+
+ /** Initializes the function's state without opening it. */
+ public void initializeFunctionWithState() throws Exception {
functionInitializationContext.getOperatorStateStore().revertToLastSuccessCheckpoint();
function = new CollectSinkFunction<>(serializer, maxBytesPerBatch,
ACCUMULATOR_NAME);
function.setRuntimeContext(runtimeContext);
function.setOperatorEventGateway(gateway);
function.initializeState(functionInitializationContext);
- function.open(DefaultOpenContext.INSTANCE);
- coordinator.handleEventFromOperator(0, 0, gateway.getNextEvent());
}
public void invoke(IN record) throws Exception {
