You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "jubins (via GitHub)" <gi...@apache.org> on 2026/06/05 20:45:21 UTC
[PR] [FLINK-39873][streaming] Add null check for connectedSocket in `SocketStreamIterator.notifyOfError()` [flink]
jubins opened a new pull request, #28332:
URL: https://github.com/apache/flink/pull/28332
## What is the purpose of the change
Fixes FLINK-39873 — The `SocketStreamIterator.notifyOfError()` method attempts to close `connectedSocket` without checking if it's null first, which can cause a `NullPointerException`.
The `connectedSocket` field is only initialized during the first call to `readNextFromStream()`. If `notifyOfError()` is called before any data is read (e.g., during early initialization errors), `connectedSocket` will be null, causing a `NullPointerException`. While the exception is currently caught and ignored, this creates unnecessary exceptions and represents poor coding practice.
This PR adds a null check before attempting to close `connectedSocket`, aligning the error handling pattern with the existing `close()` method in the same class.
## Brief change log
- Added null check for `connectedSocket` in `SocketStreamIterator.notifyOfError()` before calling `close()` method
- Ensures consistency with the null-checking pattern already used in the `close()` method (line 114)
## Verifying this change
This change is a trivial rework / code cleanup without any test coverage.
The fix follows the same defensive coding pattern already established in the `close()` method within the same class, which properly checks `if (connectedSocket != null)` before attempting to close the socket.
## Does this pull request potentially affect one of the following parts
- Dependencies (does it add or upgrade a dependency): **no**
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: **no** — `SocketStreamIterator` is annotated with `@Experimental`
- The serializers: **no**
- The runtime per-record code paths (performance sensitive): **no**
- Anything that affects deployment or recovery (JobManager, Checkpointing, Kubernetes/Yarn, ZooKeeper): **no**
- The S3 file system connector: **no**
## Documentation
- Does this pull request introduce a new feature? **no**
- If yes, how is the feature documented? **not applicable**
## Was generative AI tooling used to co-author this PR?
- [x] Yes — Claude Code was used as a pair-programming assistant. All code was written, understood, and verified by the author.
**Generated-by:** Claude Opus 4.8
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39873][streaming] Add null check for `connectedSocket` in `SocketStreamIterator.notifyOfError()` [flink]
Posted by "jubins (via GitHub)" <gi...@apache.org>.
jubins commented on PR #28332:
URL: https://github.com/apache/flink/pull/28332#issuecomment-4674179379
@MartijnVisser @RocMarshal would you be able to help final review/merge this PR?
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39873][streaming] Add null check for `connectedSocket` in `SocketStreamIterator.notifyOfError()` [flink]
Posted by "jubins (via GitHub)" <gi...@apache.org>.
jubins commented on code in PR #28332:
URL: https://github.com/apache/flink/pull/28332#discussion_r3425708337
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/experimental/SocketStreamIteratorTest.java:
##########
@@ -104,4 +104,45 @@ public void run() {
.isInstanceOf(RuntimeException.class)
.hasMessageContaining("test");
}
+
+ @Test
+ void testNotifyOfErrorBeforeConnection() throws Exception {
+ // Test that notifyOfError() can be called before connectedSocket is initialized
+ // (i.e., before any data has been read)
+ final SocketStreamIterator<Long> iterator =
+ new SocketStreamIterator<>(LongSerializer.INSTANCE);
+
+ // Call notifyOfError before any connection is established
+ // This should not throw NullPointerException
+ iterator.notifyOfError(new Exception("early error"));
+
+ // Verify that the iterator properly propagates the error
+ assertThatThrownBy(iterator::hasNext)
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("early error");
+
+ // Clean up
+ iterator.close();
+ }
+
+ @Test
+ void testNotifyOfErrorMultipleTimes() throws Exception {
+ // Test that notifyOfError() handles multiple calls gracefully
+ final SocketStreamIterator<Long> iterator =
+ new SocketStreamIterator<>(LongSerializer.INSTANCE);
+
+ // First error should be recorded
+ iterator.notifyOfError(new Exception("first error"));
+
+ // Second error should be ignored (first error takes precedence)
+ iterator.notifyOfError(new Exception("second error"));
Review Comment:
good point, added `.hasMessageNotContaining("second error")` to the assertion chain in `testNotifyOfErrorMultipleTimes`. Now the test explicitly verifies both that the first error is propagated and that the second error did not overwrite it
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39873][streaming] Add null check for `connectedSocket` in `SocketStreamIterator.notifyOfError()` [flink]
Posted by "jubins (via GitHub)" <gi...@apache.org>.
jubins commented on PR #28332:
URL: https://github.com/apache/flink/pull/28332#issuecomment-4775695903
@raminqaf @RocMarshal @snuyanzin could you please help with final review/merge if changes look good?
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39873][streaming] Add null check for `connectedSocket` in `SocketStreamIterator.notifyOfError()` [flink]
Posted by "jubins (via GitHub)" <gi...@apache.org>.
jubins commented on PR #28332:
URL: https://github.com/apache/flink/pull/28332#issuecomment-4641466317
@spuru9 Thanks for the review! Do you know who could help get this merged?
@snuyanzin Would you be able to help with the merge?
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39873][streaming] Add null check for `connectedSocket` in `SocketStreamIterator.notifyOfError()` [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #28332:
URL: https://github.com/apache/flink/pull/28332#issuecomment-4635437144
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "a355be230869f2c271920d9c8e0a37e9ab41831f",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "a355be230869f2c271920d9c8e0a37e9ab41831f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* a355be230869f2c271920d9c8e0a37e9ab41831f UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39873][streaming] Add null check for `connectedSocket` in `SocketStreamIterator.notifyOfError()` [flink]
Posted by "jubins (via GitHub)" <gi...@apache.org>.
jubins commented on PR #28332:
URL: https://github.com/apache/flink/pull/28332#issuecomment-4827262990
@snuyanzin you're right that the tests passed without the fix, the original catch (Throwable ignored) was silently swallowing the `NullPointerException` from `connectedSocket.close()`. I've fixed this PR with two improvements: the `null` check itself, and replacing the overly broad `catch (Throwable)` with the precise catch `(IOException)`. Let me know if anything else.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39873][streaming] Add null check for `connectedSocket` in `SocketStreamIterator.notifyOfError()` [flink]
Posted by "jubins (via GitHub)" <gi...@apache.org>.
jubins commented on PR #28332:
URL: https://github.com/apache/flink/pull/28332#issuecomment-4747871014
@raminqaf let me know what you think and if you have further comments
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39873][streaming] Add null check for `connectedSocket` in `SocketStreamIterator.notifyOfError()` [flink]
Posted by "raminqaf (via GitHub)" <gi...@apache.org>.
raminqaf commented on code in PR #28332:
URL: https://github.com/apache/flink/pull/28332#discussion_r3413254653
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/experimental/SocketStreamIteratorTest.java:
##########
@@ -104,4 +104,45 @@ public void run() {
.isInstanceOf(RuntimeException.class)
.hasMessageContaining("test");
}
+
+ @Test
+ void testNotifyOfErrorBeforeConnection() throws Exception {
+ // Test that notifyOfError() can be called before connectedSocket is initialized
+ // (i.e., before any data has been read)
+ final SocketStreamIterator<Long> iterator =
+ new SocketStreamIterator<>(LongSerializer.INSTANCE);
+
+ // Call notifyOfError before any connection is established
+ // This should not throw NullPointerException
+ iterator.notifyOfError(new Exception("early error"));
+
+ // Verify that the iterator properly propagates the error
+ assertThatThrownBy(iterator::hasNext)
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("early error");
+
+ // Clean up
+ iterator.close();
+ }
+
+ @Test
+ void testNotifyOfErrorMultipleTimes() throws Exception {
+ // Test that notifyOfError() handles multiple calls gracefully
+ final SocketStreamIterator<Long> iterator =
+ new SocketStreamIterator<>(LongSerializer.INSTANCE);
+
+ // First error should be recorded
+ iterator.notifyOfError(new Exception("first error"));
+
+ // Second error should be ignored (first error takes precedence)
+ iterator.notifyOfError(new Exception("second error"));
Review Comment:
Could you elaborate how the second error is ignored? I see that you have asserted the first error but no assertion is done on the second error
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39873][streaming] Add null check for `connectedSocket` in `SocketStreamIterator.notifyOfError()` [flink]
Posted by "raminqaf (via GitHub)" <gi...@apache.org>.
raminqaf commented on code in PR #28332:
URL: https://github.com/apache/flink/pull/28332#discussion_r3413237862
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/experimental/SocketStreamIteratorTest.java:
##########
@@ -104,4 +104,45 @@ public void run() {
.isInstanceOf(RuntimeException.class)
.hasMessageContaining("test");
}
+
+ @Test
+ void testNotifyOfErrorBeforeConnection() throws Exception {
+ // Test that notifyOfError() can be called before connectedSocket is initialized
+ // (i.e., before any data has been read)
Review Comment:
Do we need this comments? Seems a bit duplicated. The name of the test already covers it
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39873][streaming] Add null check for `connectedSocket` in `SocketStreamIterator.notifyOfError()` [flink]
Posted by "jubins (via GitHub)" <gi...@apache.org>.
jubins commented on code in PR #28332:
URL: https://github.com/apache/flink/pull/28332#discussion_r3425709842
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/experimental/SocketStreamIteratorTest.java:
##########
@@ -104,4 +104,45 @@ public void run() {
.isInstanceOf(RuntimeException.class)
.hasMessageContaining("test");
}
+
+ @Test
+ void testNotifyOfErrorBeforeConnection() throws Exception {
+ // Test that notifyOfError() can be called before connectedSocket is initialized
+ // (i.e., before any data has been read)
Review Comment:
removed some redundant comments
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39873][streaming] Add null check for `connectedSocket` in `SocketStreamIterator.notifyOfError()` [flink]
Posted by "jubins (via GitHub)" <gi...@apache.org>.
jubins commented on PR #28332:
URL: https://github.com/apache/flink/pull/28332#issuecomment-4693780382
hi @snuyanzin @RocMarshal can I get a final review/approval/merge on this PR? thank you!
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org