Hi, I use Rebus for integration events. I have to switch to Kafka for some reason and wanted to use this library.
Unfortunately, I have encountered a problem.

By adding more topics, the consumer subscribes only to the first one,
so that, although many topics are sent, only one is received (subscribed to).

The current setup works flawless for RabbitMQ or InMemory transport implementations.

I suspect that the issue may be related to topic subscription.

I have reproduced the issue on your example:

Steps to reproduce:

Using the Examples projects available: Scaleout.Producer, Scaleout.Messages and Scaleout.Consumers,
I modified the Producer and Consumers project as follows:

• I added TestMessage2 (a copy of TestMessge).
• I added TestMessageHandler2 : IHandleMessages<TestMessage2> (copy of TestMessageHandler).
• I registered the new handler (Consumers).
• I added Rebus Routing for Assembly TestMessage2 (Producer and Consumers).
• I subscribed to both topics: (Consumers)
  bus.Subscribe<TestMessage>().Wait();
bus.Subscribe<TestMessage2>().Wait();
• I published messages from both TestMessage and TestMessage2 to the do-while loop in Producer.

Expected behaviour:
Console outputs messages from both topics.

Actual:
Console outputs only topics from TestMessage.

Any idea what could be the cause, or maybe I'm doing something wrong?

Hi,
trying to get started. I want to be able to produce to two topics. Both topics have with different values. The values need to be serialized with protobuf.

var eventProvider1 =  new ProducerBuilder<string, Event1>(producerConfig)
    .SetValueSerializer(....)
    .Build()
var eventProvider2 =  new ProducerBuilder<string, Event2>(producerConfig)
    .SetValueSerializer(....)
    .Build()

Is this something that can be achieved? I saw that the KafkaTransport has a Producer<string, byte[]>

How configure the Transport to work with Partitions? Are there any special configurations?

I'm playing around with the Kafka transport to examine its viability. When I stand up a basic Kafka environment on docker (2.1 Kafka)...no tweaks...just a single broker and single zookeeper, The throughput I'm seeing is 10K msg per second. Knowing that a fully tweaked kafka can handle up to 2 million. Is the 10K/s benchmark indicative of what others have experienced using this connector? Just trying to gauge whether the bottleneck is the connector, the confluent library, or the fact that I have a generic kafka environment. Whats the throughput others have experienced?

Hello,

Does the saga pattern implemented in Rebus and Kafka go together?

I'm very happy to see that you have coded a transport layer for Kafka. However it's not part of the rebus transports available on rebus github. I've on Stackoverflow found this answer for a question regarding Kafka and the saga pattern: https://stackoverflow.com/a/58748235

I case your answer is yes, please explain the conditions and setup of Rebus. This may be among a lot of other setups Kafka partitions and topics.

Regards
Jan Rou

I noticed that the SimpleRetryStrategy doesn't work with the Rebus.Kafka plugin. When I have a message handler that throws an exception, Rebus will log it as a warning, but the Rebus.Kafka just goes on its merry way and consumes the next message in the kafka topic as if the previous message was processed successfully. I can see where it can be difficult to handle replaying of a previous message as it requires moving the offset backwards to try reprocessing it before finally giving up and forwarding the message to an actual deadletter topic. Is there something that I'm missing that needs to be configured to turn on the retry and deadletter mechanism or is this feature not implemented in the plugin yet?

It seems that we always have the option to build out a retry mechanism and forwarding of the message to an error queue from within handler code, but this is not ideal. It leads to a lot of repetitive code and is not safe from the perspective that the message may never get completely processed if the service is restarted as the message is in the middle of its retry logic. It would be nice if that was leveraging the fallback mechanism already built into rebus as it accounts for those scenarios.

One option I can think of is to block performing a commit() until after the successful processing of a message by the handler (this would obviously require turning off the autocommit feature). If an exception is thrown., don't allow the code to determine if a commit() should be called. Instead, try calling the handler again. Once the retry is exceeded, push the message to the deadletter topic defined in the SimpleRetryStrategy and then we can proceed to the commit() logic and subsequent consume() of the next message.

It seems like this would be the most straight forward way to hook into the SimpleRetryStrategy. Obviously, if SimpleRetry is not defined, the current implementation seems like it would be appropriate.

Is there a setting I'm missing somewhere or is this a legitimate limitation of the plugin?

I am trying to use Rebus.Kafka in .Net core Web API. I tried to do by register it in startup.cs but it always throw "The transport initialization procedure is still incomplete. There is no confirmation of completion of the subscription to the input queue".

Code is like this :
services.AddRebus(configure => configure
.Transport(t => t.UseKafka("localhost:9092", "MainQueue",producerConfig,consumerConfig))
.Routing(r =>
{
r.TypeBased()
.MapAssemblyOf(MainQueue)
})

Hi there, just want to direct your attention towards this question 👇

https://stackoverflow.com/questions/60834095/manually-commiting-in-rebus-with-kafka

🙂

It seems that using Confluent 1.5.3+ causes the auto topic creation capability to fail. This seems to result in stalling the Rebus.Kafka connector to the point where calling Publish() through Rebus will eventually result in a large delay followed by an error condition.

I've tried setting the AutoCreateTopicsEnabled flag in the Confluent.Consumer structure to TRUE. Plus, I verified that the kafka brokers can also accept Auto creation of topics. It just plain doesn't work. Many people are complaining about this very problem in version 1.5.3 and higher.

Recompiling Rebus.Kafka with 1.4.4 of Confluent's library seems to straighten everything out. According to Confluent, they turned off Auto Topic creation by default in the 1.5.3. But, apparently there is no way I've found to override that and turn it back on.

Auto topic creation is not a problem for production deploys since we want to manually create/adjust topics...but when it comes to developer workflows on isolated dev boxes, forcing devs to create topics seems like a step too far....the software should just do that as a reaction to working with Kafka.

Not sure if the solution is to go back to 1.4.4 in the nuget references of Rebus.Kafka...or just track that this is going to be an issue for people to be aware of until Confluent puts out a release that doesn't exhibit this problem.

Hello,

looking for a messaging solution that supports kafka and outbox. Is this the right addon?

Thanks.