rocklib / rocklib.messaging Goto Github PK

CloudEvents is currently at version 1.0.2: https://github.com/cloudevents/spec.

Currently, the library will only validate events that specify 1.0 and will error for any other specification version.

Spec versions 1.0.1 and 1.0.2 are backwards compatible, so it would be great if the library could either support these or relax its validation to treat them as 1.0.

https://github.com/RockLib/RockLib.Messaging/blob/main/RockLib.Messaging.Kafka/KafkaSender.cs#L215

IProducer<T1,T2> implements IDisposable and should also be disposed.

Is there a reason that we are using an older dependency of Confluent.Kafka (1.8.2). Some of my teams were being advised to use Confluent.Kafka natively instead of RockLib and wanted to get a better understanding if there was any specific reason we are still using 1.8.2.

Problem
Whenever the configuration events occur, the reloading config creates a new service and disposes of the original. Coupled with the fact that disposing some services, disposes of their dependencies, regardless of origin, this can result in disposal of the base dependency, needed for the newly created service.

Example:

builder.Services.Configure<SQSSenderOptions>(QueueDefaults.Name, builder.Configuration.GetSection("Queue:Event"));
builder.Services.TryAddAWSService<IAmazonSQS>();
builder.Services.AddSQSSender(QueueDefaults.Name);

With this example, the base AmazonSQSClient will be disposed upon the first configuration event received.

Workaround
Do not use the reloading config.

builder.Services.AddSQSSender(QueueDefaults.Name, reloadOnChange: false);

Note that this workaround does not account for any cases in which a sender service is disposed of, as it will also result in the disposal of the base amazon client.

Possible solutions

• The library can properly support DI scoping.
• The services should never dispose anything they did not explicitly instantiate themselves, unless they are told to do so.

Using AddKafkaSender extension function from KafkaExtensions to instantiate the KafkaSender doesn't read the SchemaId property from the configureOptions. This leads to a problem of not setting the SchemaId value to the KafkaSender instance.

Having the ability to pause a receiver will help when creating circuit breakers on dependencies.

Something like https://medium.com/@vaibhav.vb24/kafka-confluent-pause-and-resume-consumer-cda7305944bf

KafakaReceiverMessage implements ReceiverMessage, and ReceiverMessage has an SemaphoreSlim member. SemaphoreSlim implements IDisposable, but ReceiverMessage does not implement IDisposable so the SemaphoreSlim does not get disposed. This leaves a potential performance issue that could lead to exceptions or necessary garbage collection. cc @JustinMoss

Originally posted by @peteraritchie in #120 (comment)

_disposeSource is also disposable, the Cancel method does not dispose of the object and KafkaReceiver.Dispose should dispose of it.