Fixed an issue in InputStreamSubscriber that could cause NPE to be thrown when close and o…

[zoewangg]

…nSubscribed get invoked concurrently.

Motivation and Context

Fix #4081 The root cause is a race condition that onSubscribe and close are invoked concurrently, the subscription could be null, causing NPE to be thrown from close method.

Modifications

Add a lock in onSubscribe and close

Testing

Tested it locally and verified the test no longer failed with this change. This test is not checked in because it cannot reproduce this issue reliably.

    @Test
    public void closeOnSubscribe_invokeAtSameTime_shouldNotFail() {
        int count = 0;
        for (int i = 0; i < 10000; i++) {
            CompletableFuture<Void> future = CompletableFuture.runAsync(() -> subscriber.close());
            //publisher.subscribe(subscriber);
            subscriber.onSubscribe(new Subscription() {
                @Override
                public void request(long n) {

                }

                @Override
                public void cancel() {

                }
            });
            future.join();
        }

    }

Screenshots (if appropriate)

Types of changes

• Bug fix (non-breaking change which fixes an issue)
• New feature (non-breaking change which adds functionality)

Checklist

• I have read the CONTRIBUTING document
• Local run of mvn install succeeds
• My code follows the code style of this project
• My change requires a change to the Javadoc documentation
• I have updated the Javadoc documentation accordingly
• I have added tests to cover my changes
• All new and existing tests passed
• I have added a changelog entry. Adding a new entry must be accomplished by running the scripts/new-change script and following the instructions. Commit the new file created by the script in .changes/next-release with your changes.
• My change is to implement 1.11 parity feature and I have updated LaunchChangelog

License

• I confirm that this pull request can be released under the Apache 2 license

[sonarcloud]

Kudos, SonarCloud Quality Gate passed!   

0 Bugs
0 Vulnerabilities
0 Security Hotspots
3 Code Smells

91.7% Coverage
0.0% Duplication