NOISSUE - Kafka support

[rodneyosodo]

What does this do?

Adds Kafka support. Since we are introducing a multi-broker system, we need to add support for Kafka. This allows us to use Kafka as a message broker for our system. Kafka has been chosen because it is the most popular legacy message broker. This allows us to support legacy systems that use Kafka.

Which issue(s) does this PR fix/relate to?

Resolves #990

List any changes that modify/break current functionality

N/A

Have you included tests for your changes?

Yes

Did you document any new/modified functionality?

absmach/supermq-docs#157

Notes

The publisher implementation is based on the segmentio/kafka-go library. Publishing messages is well supported by the library, but subscribing to topics is not. The library does not provide a way to subscribe to all topics, but only to a specific topic. This is a problem because the Mainflux platform uses a topic per channel, and the number of channels is not known in advance. The solution is to use the Zookeeper library to get a list of all topics and then subscribe to each of them. The list of topics is obtained by connecting to the Zookeeper server and reading the list of topics from the /brokers/topics node. The first message received from the topic can be lost if subscription happens closely followed by publishing. After the subscription, we guarantee that all messages will be received.

[codecov-commenter]

Codecov Report

Merging #1579 (6a1e87a) into master (28f4965) will increase coverage by 0.07%.
Report is 28 commits behind head on master.
The diff coverage is 70.91%.

❗ Current head 6a1e87a differs from pull request most recent head 8537a3c. Consider uploading reports for the commit 8537a3c to get more accurate results

@@            Coverage Diff             @@
##           master    #1579      +/-   ##
==========================================
+ Coverage   67.14%   67.21%   +0.07%     
==========================================
  Files         120      122       +2     
  Lines        9145     9341     +196     
==========================================
+ Hits         6140     6279     +139     
- Misses       2368     2413      +45     
- Partials      637      649      +12     

Files	Coverage Δ
pkg/messaging/kafka/publisher.go	63.79% <63.79%> (ø)
pkg/messaging/kafka/pubsub.go	73.91% <73.91%> (ø)

📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more

[drasko]

I would construct this one as chansPrefix + ".>"

[rodneyosodo]

That is what is implemented currently across all brokers. Should we change?

[dborovcanin]

@0x6f736f646f Do not forget to update Makefile and docker-compose.yml to enable run with MF_BROKER_TYPE=kafka make run.

[dborovcanin]

Please fix CI.

[dborovcanin]

CI is failing.

[rodneyosodo]

The LeaderNotAvailable is retriable according to this docs. A possible workaround would be to retry and if not possibly fail and manage the fails. The library we are using retries up to 1s then fails. As demonstrated earlier, if you retry for up to at least 10 seconds the writer will write the message. The current work around it is to create topics using the Kafka command inside the broker as referenced here.
Our problem is mainly on the CI as when mainflux is deployed you can be able to publish a message from one adapter and receive it using another. The interim workaround is to create the topics first before testing the code.

Another problem is that on mainflux when we publish message for the first time it throws the same error mainflux-http | {"level":"warn","message":"Method publish to channel 242bcd02-fb24-4c8c-b56d-de97ad56de34 took 49.085996ms to complete with error: [5] Leader Not Available: the cluster is in the middle of a leadership election and there is currently no leader for this partition and hence it is unavailable for writes.","ts":"2022-06-27T14:01:40.789712934Z"}

This is an error with regard to Kafka as it is being experienced with other library users such as Sarama.

My recommendation would be to increase the retrying on public method upto 10s then manage the error after that

[dborovcanin]

The LeaderNotAvailable is retriable according to this docs

Can you point out the exact section for this?

Our problem is mainly on the CI as when mainflux is deployed you can be able to publish a message from one adapter and receive it using another. The interim workaround is to create the topics first before testing the code.

Looks like the problem is actually that first-time writing always arises this error, not due to Kafka not being ready but because we do not have a topic created. I guess Kafka creates a topic internally when we try to publish, and the next attempt will pass because on the first - Kafka took care of topic for us. Please investigate if it works like that and how to correctly handle topics lifetime (we do not remove them when we unsubscribe, and I assume there are other ways to configure Kafka to handle it for us).

[rodneyosodo]

Can you point out the exact section for this?

You can check it from here

Please investigate if it works like

Yes it works like that.
I have tried two approaches:

1. To create the topic as reference here but only when KAFKA_AUTO_CREATE_TOPICS_ENABLE='false'

   This doesn't work as expected as

topicConfigs := []kafka.TopicConfig{
	{
		Topic:             subject,
		NumPartitions:     1,
		ReplicationFactor: 1,
	},
}
err = pub.conn.CreateTopics(topicConfigs...)
if err != nil {
	return err
}

after initially creating the kafka connection object.

3. To create the topic while publishing as reference here
   This one worked fine.

[rodneyosodo]

Topic management

Kafka topics are the categories used to organize messages. Each topic has a name that is unique across the entire Kafka cluster.

Creation

• We are manually creating the topics before publishing to it.
• If we try and create an already created topic the topic it is idempotent since the values haven't changed

Deletion

• Since we are creating the topics ourselves we have the privilege to delete them ourselves. This is not an elegant solution but a feasible solution to our case.
• Before we can delete any topic we would want to make sure the deletion of topics is enabled in our cluster. Set delete.topic.enable=true.
• We have to make sure all consumers have stopped consuming the data from the topic we want to delete, otherwise, kafka will throw errors.
• If there is a chance that a consumer is up and running, the topic will get auto-created if the cluster-wide property auto.create.topics.enable=true. Not bad per se, but it will use a default number of partitions (1) and a replication factor (1), which might not be what we wanted.

We are currently deleting topics when we close the publisher interface

Publishing message

• Configured retries to be at least 1000 since we care about data delivery to clients Ref
• Configured producer for to wait for acknowledgment Ref
• Stuck to random partitioning when writing to topics Ref

[drasko]

LGTM

[rodneyosodo]

From this it appears that in the segmentio library you can list all topics with a bit of work around. But the problem is when the MQTT forwarder is subscribing to the broker, it uses the wildcard to subscribe to all topics but they are no topics in the kakfa broker.

[rodneyosodo]

Regarding Kafka support for wildcard, I have tested using confluent-kafka-python and it supports. I think the issue is the library we are using I have tried to pass in a regex on the topic and it throws this error

2023/02/07 14:24:37 Failed to join group groupID2: [17] Invalid Topic: a request which attempted to access an invalid topic (e.g. one which has an illegal name), or if an attempt was made to write to an internal topic (such as the consumer offsets topic)
2023/02/07 14:24:37 Leaving group groupID2, member main@r0x6f736f646f (github.com/segmentio/kafka-go)-77b03394-27c9-4846-89c2-fe1fb51f7034

I have tried with another library https://github.com/confluentinc/confluent-kafka-go and it works. I want to write the implementation on this kafka PR and update after testing it on mainflux

[dborovcanin]

Let's address the subscriber problem and research our options in a separate PR and make this PR only a Publisher implementation. Using Subscriber will simply panic. Don't forget to document it.

[dborovcanin]

This PR might be a better fit for https://github.com/absmach/mg-contrib since Kafka does not support all the features MG messaging requires, but may also be useful for integration with systems that use Kafka and do not need those features.

[dborovcanin]

As already commented here, this may be a better fit for the mg-contrib repository. I'm closing this PR.