[FLINK-39873][streaming] Add null check for connectedSocket in SocketStreamIterator.notifyOfError()

[jubins]

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)
• Added two new test cases in SocketStreamIteratorTest:
  • testNotifyOfErrorBeforeConnection() — verifies that notifyOfError() can be called safely before connectedSocket is initialized
  • testNotifyOfErrorMultipleTimes() — ensures multiple error notifications are handled gracefully with the first error taking precedence

Verifying this change

This change is covered by two new unit tests in SocketStreamIteratorTest:

• testNotifyOfErrorBeforeConnection() — Verifies that calling notifyOfError() before any connection is established (i.e., when connectedSocket is null) does not throw a NullPointerException and properly propagates the error to hasNext()
• testNotifyOfErrorMultipleTimes() — Validates that multiple calls to notifyOfError() are handled correctly, with the first error taking precedence (as per the existing logic in the method)

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?

• 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

[flinkbot]

CI report:

• e280228 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build