Graceful handling for PulsarClient.CloseAsync

[LeoSht]

Hi, Current version of the client appears to rely on socket connection dispose call to break out of read operation when PulsarClient.CloseAsync is called. This leads to warnings in the log and prints out an exception like this: System.ObjectDisposedException: Cannot access a disposed object. Object name: 'SslStream'. ...

These exceptions show up a few seconds after the PulsarClient.CloseAsync call was made, so don't easily show up on integration tests.

The proposed change supplies a CancellationToken to the PipeReader.ReadAsync method. And then triggers token cancellation on Dispose method call in ClientCnx.

Here is an example test (C#) to observe the fix, assuming we have an ILogger implementation that collects log statements as a string:

    [Fact]
    public async Task Client_Close_Graceful()
    {
        // Arrange
        var logTextBuilder = new StringBuilder();
        var logger = new StringLogger(logTextBuilder, LogLevel.Information);

        // Act
        var client = await GetClientAsync(logger);
        var topic = "public/default/test_graceful_client_close";
        var producer = await client.NewProducer()
            .Topic(topic)
            .CreateAsync();
        await producer.DisposeAsync();
        await client.CloseAsync();

        // Wait for socket timeout
        await Task.Delay(TimeSpan.FromSeconds(60));

        // Assert
        var logOutput = logger.Output.ToString();
        Assert.DoesNotContain("Exception", logOutput);
    }

Let me know if you need any additional information. Regards, Leo.

[Lanayx]

Published 2.15.1

[LeoSht]

Thank you!