Dispose method for KafkaStream object throws 'System.ObjectDisposedException: handle is destroyed'

[dzmitry-horbar]

Description

Hello, I have created simple console application to test GlobalTable + InMemory state store and everything works fine untill we close application. As recommended I do a call to KafkaStream.Dispose() and it throws an error :

Can you please help me with that ? What could be the cause of this issue ?

How to reproduce

Create .net 5.0 console application with Streamiz.Kafka.Net v1.1.5 I removed reading from state store because it doesn't affect this issue, the minimal code to reproduce it below.

Configure and start:

            var config = new StreamConfig<StringSerDes, StringSerDes>();
            config.ApplicationId = "test-app-123";
            config.BootstrapServers = "localhost:9092";
            config.AutoOffsetReset = AutoOffsetReset.Earliest;

            StreamBuilder builder = new StreamBuilder();
            builder.GlobalTable("dima-test", InMemory<string, string>.As("dima-test-store"));

            Topology t = builder.Build();
            KafkaStream stream = new KafkaStream(t, config);
            await stream.StartAsync();

            await Task.Delay(2000);

            stream.Dispose();

Checklist

Please provide the following information:

• [x] A complete (i.e. we can run it), minimal program demonstrating the problem. No need to supply a project file.
• [x] A code snippet with your topology builder (ex: builder.Stream<string, string>("topic").to("an-another-topic");)
• [x] Streamiz.Kafka.Net nuget version. - v1.1.5
• [x] Apache Kafka version. - v1.6.2 (which is referenced from Streamiz library)
• [x] Client configuration.
• [x] Operating system. - Windows 10
• [x] Provide logs (with in debug mode (log4net and StreamConfig.Debug) as necessary in configuration).
• [ ] Critical issue. - no

[LGouellec]

Hi @dzmitry-horbar,

I will check ASAP ! Thanks for your issue.

Regards,

[dzmitry-horbar]

Hi @LGouellec , not sure should I create a new issue or could report it here, but when I tried to update Confluent.Kafka to latest version v1.7.0 and it cause KafkaStreams to fail on start with following exception:

Stack trace:

[LGouellec]

Hi @LGouellec , not sure should I create a new issue or could report it here, but when I tried to update Confluent.Kafka to latest version v1.7.0 and it cause KafkaStreams to fail on start with following exception:

Stack trace:

KafkaStreams use Confluent.Kafka 1.6.2, and in 1.7.0 some breaking changes are appeared (like some method change, same issue as above). In next release, my library will use 1.7.0 but need some changes in my code.

[LGouellec]

Description

Hello, I have created simple console application to test GlobalTable + InMemory state store and everything works fine untill we close application. As recommended I do a call to KafkaStream.Dispose() and it throws an error :

Can you please help me with that ? What could be the cause of this issue ?

How to reproduce

Create .net 5.0 console application with Streamiz.Kafka.Net v1.1.5 I removed reading from state store because it doesn't affect this issue, the minimal code to reproduce it below.

Configure and start:

            var config = new StreamConfig<StringSerDes, StringSerDes>();
            config.ApplicationId = "test-app-123";
            config.BootstrapServers = "localhost:9092";
            config.AutoOffsetReset = AutoOffsetReset.Earliest;

            StreamBuilder builder = new StreamBuilder();
            builder.GlobalTable("dima-test", InMemory<string, string>.As("dima-test-store"));

            Topology t = builder.Build();
            KafkaStream stream = new KafkaStream(t, config);
            await stream.StartAsync();

            await Task.Delay(2000);

            stream.Dispose();

Checklist

Please provide the following information:

• [x] A complete (i.e. we can run it), minimal program demonstrating the problem. No need to supply a project file.
• [x] A code snippet with your topology builder (ex: builder.Stream<string, string>("topic").to("an-another-topic");)
• [x] Streamiz.Kafka.Net nuget version. - v1.1.5
• [x] Apache Kafka version. - v1.6.2 (which is referenced from Streamiz library)
• [x] Client configuration.
• [x] Operating system. - Windows 10
• [x] Provide logs (with in debug mode (log4net and StreamConfig.Debug) as necessary in configuration).
• [ ] Critical issue. - no

I reproduced your fix, and in fact, sometimes an error is logged with ObjectDisposedException raised. It's not a fatal error because it's catch.

But I fixed this issue in my branch fix/84, will release in next release 1.2.0 in some weeks I hope.

[dzmitry-horbar]

Good work, thanks for quick turn a round ! :)

[LGouellec]

1.2.0 released