Hono: Support Apache Kafka as alternative to AMQP 1.0 Messaging Network

[sophokles73]

Hono does not strive to implement a new message broker but instead tries to leverage existing messaging infrastructure to meet its quality goals. It seems reasonable to assume that the amount of telemetry data flowing upstream from devices to back end applications will be orders of magnitude larger than the data being sent to devices from back end applications in order to invoke operations or configure properties on the devices.

It seems therefore feasible to leverage Apache Kafka (at least) for transporting telemetry data downstream taking advantage of Kafka's qualities regarding horizontal scalability, low latency and fault tolerance.

We will need to come up with ideas of how to map downstream data to Kafka's _topics_ and how to allow for parallel message production/consumption in the context of Hono.

[ppatierno]

We have an important point : AMQP 1.0
I think that it's a very full featured protocol to use for Hono architecture and I agree on using it.
It's also true that Apache Kafka is a great ingestion system for an huge amount of data (at high rate) and so it's useful for telemetry. The bad point is that Apache Kafka doesn't support AMQP 1.0 but it's based on a proprietary protocol.
We could lost performance on telemetry, if we put a layer before Kafka for ingesting data using AMQP 1.0 and then route data to Kafka with its proprietary protocol.

[sophokles73]

That's true, converting from AMQP 1.0 messages to a Kafka producer will cost some cycles. However, my expectation would be that this will be made up by the fact that Kafka can ingest and deliver the messages with a high degree of parallelism. So this might introduce some latency to the transmission of an individual message but will (hopefully) still improve the throughput of the overall system.

This, of course, requires that we run multiple Hono endpoints that can serve as parallel Kafka producers ...

[ppatierno]

Considering that Kafka doesn't support AMQP 1.0 natively we can think it as a service for our IoT Connector. Of course, it won't be the primary ingestion system but a service which connects to the IoT Connector to receive data via AMQP 1.0 and provides them in a Kafka fashion to other services like for example Apache Storm.

[ppatierno]

We could try to use this AMQP Kafka bridge (https://github.com/rhiot/amqp-kafka-bridge) for this issue.

[danielbohn]

Hi i would like to contribute more in the Hono project. I work in my freetime a lot with kafka. So i am interested to have a deeper look if it is possible to use kafka as a underlying messaging infrastructure for hono. I would start to look at this AMQP Kafka bridge.

[sophokles73]

Great :+1:

[ppatierno]

@danielbohn that's great ! Consider it as a POC for now and that the Vert.x community is working on a Vert.x client for Kafka so it could be possible to use it inside the bridge instead using directly the official Kafka client.

[danielbohn]

@sophokles73 and @ppatierno i have an example with Kafka and the AMQP-Kafka bridge running on my local machine. I would now integrate the Kafka messaging infrastructure in the hono architecture behind the Qpid dispatch router for the telemetry endpoint.

[danielbohn]

@ppatierno is a docker image für the Kafka-AMQP bridge available ?

[sophokles73]

@danielbohn ,

this sounds interesting but I wonder what that would actually entail. Is this merely a config file for dispatch router?

[danielbohn]

@sophokles73 yes it would be a config file for the dispatch router and a new docker-compose file in the example folder. Later on we can also use the Kafka messaging infrastructure for the event endpoint.

[ppatierno]

@sophokles73 @danielbohn on the top of my head should be enabling an outgoing connection from the router to the AMQP bridge (which listens as a server).

@danielbohn there is no Docker image available online on Docker Hub but there is the Docker file for building that. In order to avoid a waste of time for you let me try the connection from router to bridge in a configuration outside of Hono scenario. I'll come back to you soon. ;)

[danielbohn]

@ppatierno could you please explain "on the top of my head should be enabling an outgoing connection ..." a little bit more. I have a example running with your bridge for sending and receiving messages via AMQP 1.0 and kafka

[sophokles73]

Before we spend more effort on this I would like to better understand what we are trying to achieve here.
I am also not sure if we would consider the Kafka broker as part of a Hono instance in the same way as we would consider an AMQP 1.0 based message broker part of Hono.

My understanding is that we can easily pipe messages into a downstream Kafka server from Dipatch Router. But downstream receivers would then directly attach to Kafka in order to process the messages, or wouldn't they?

[ppatierno]

Using the bridge you should be able to put message to Kafka using AMQP 1.0 so through the router network.
In the same way you can get messages from Kafka using the same AMQP bridge or even connect directly to the Kafka cluster.

[dguggemos]

@sophokles73 With this approach we would have yet another place where we need to enforce security. We already have problems to do this for two interfaces (Hono and QDR). Or do you think Kafka is not part of Hono platform in this case? I think in a shared cloud environment this is not feasible.

[sophokles73]

@dguggemos,

one could argue that Kafka is a means for (optionally) persisting telemetry data that is specific to the application. In that case I would consider it NOT part of Hono.
I share your concerns regarding having another interface where we would need to enforce security.

@ppatierno: I have a hard time imagining how data can be consumed through an AMQP 1.0 bridge given the different concepts of how a client works (Kafka vs. AMQP). But maybe you have found a good way to map AMQP to Kafka's consumer protocol ...

[ppatierno]

@sophokles73 yes it's what I tried to do with the bridge ... but feedback are welcome ... you can see more information on mapping on the project landing page and the Wiki section ;)

[amq]

Similar implementation:
https://www.confluent.io/product/confluent-open-source/
https://github.com/datamountaineer/stream-reactor (underlying MQTT and CoAP connectors)

[sophokles73]

Our friends from the Eclipse Ditto project have added support for forwarding messages received from Hono to a user provided Kafka instance. So, people who want the messages end up in a Kafka cluster should use Hono together with Ditto and follow the instructions provided by the Ditto project.

[b-abel]

In my view, replacing the AMQP messaging network with a Kafka cluster could still be an interesting option. I think it makes sense to keep the discussion in this issue, even though other goals and strategies have been discussed here.
@sophokles73, would you mind reopening the issue?

[sophokles73]

@b-abel sure

[sophokles73]

however, IMHO we should not discuss Kafka as an alternative implementation of the AMQP 1.0 Messaging Network but as an alternative to the AMQP 1.0 Messaging Network, right? If you agree, we can change the title accordingly ...

[b-abel]

however, IMHO we should not discuss Kafka as an alternative implementation of the AMQP 1.0 Messaging Network but as an alternative _to_ the AMQP 1.0 Messaging Network, right? If you agree, we can change the title accordingly ...

Absolutely!

[b-abel]

Proposal

The first step is to replace AMQP with Kafka for the downstream direction. Commands are out of scope for now.
For this, we need a Hono Client that uses Kafka instead of AMQP for (downstream) messaging.
As a prerequisite, we should refactor the Hono Client so that it no longer exposes AMQP specifics to the outside (i.e. to the protocol adapter) and can be used protocol-independently with AMQP and Kafka.

Message Ordering

Keeping the messages sent via Hono in order is an important feature, especially when it is used in combination with Ditto.
Message ordering in Kafka comes with significant operational costs:

• Performance: no parallel processing possible, both for the producer and the consumer
• since Kafka guarantees the order only for data that is located on the same topic partition, the data cannot simply be distributed evenly across partitions, which can lead to highly imbalanced disk usage and load the nodes in the Kafka cluster

Regarding the requirements, I assume that the guaranteed order per device is sufficient and does not need to be guaranteed between several devices. Or does anyone see a different use case, e.g. for gateways?
I suggest guaranteeing the order for events, but not for telemetry messages (@thjaeckle, would that work for Ditto?). So the distinction between telemetry as a cheap, fast, but not guaranteed type and more expensive events would be usefully maintained, even though, unlike with AMQP, telemetry messages are persisted with Kafka. Besides, it is questionable anyway whether a guaranteed order without a guaranteed delivery makes sense at all.

Topics

Topic names could contain message types (event or telemetry), tenant-id, and device-id as well as combinations of them.
An important decision factor for the selection is the topic-specific configuration options.

Suggestions:

1. topic per tenant: tenant isolation, allows separate users and corresponding ACLs for each tenant, when deleting the tenant you can simply remove the topic
2. topic per message type: allows to set different retention policies and thus influence the storage requirements

When creating a tenant the topics event/<tenant-id> and telemetry/<tenant-id> should be created (+ Kafka account + ACLs).

Using separate topics for tenants has an additional advantage: it automatically allows several northbound applications to consume the messages, which would be a useful new feature. These applications would probably be tenant-specific.
To include the device-id in the topic would IMHO not add any value, only cause management overhead and it would be necessary to somehow force only one partition per topic to guarantee message-ordering.

Partitions

Hono should not impose requirements regarding the number of partitions, because this should be determined by operational requirements (too few partitions slow down the consumers, because parallelization is determined by the number of partitions, too many partitions lead to significantly increasing resource consumption).

Suggestions:

1. telemetry: do not use a key (i.e. key is null). This way the messages are distributed evenly ("round-robin") over all partitions of the topic, which results in an even distribution in the cluster. Besides, you can add partitions to telemetry topics at any time, depending on operational requirements.
2. events: take the device-id as key. It uniquely identifies the device if the tenant is already part of the topic name. The key ensures that the message order of the events of a device.

NB: Adding partitions to a topic is a problem for topics where the message order matters because the distribution of keys on partitions is recalculated and messages with a certain key, may be assigned to a new partition. Therefore, the Kafka administrator must perform capacity planning in advance and perform a "manual" migration when adding a new partition to a topic.

Message ACKs

At least for telemetry messages with AT LEAST ONCE delivery, the device currently knows immediately whether a consumer is connected. In this respect, Hono's behavior changes from the point of view of the device. This is probably inevitable.

Open Questions

Message Format

Regarding the message format, it would probably be most similar to the current behavior if the message just contains the payload, and all (application) properties Hono sets now are added to the Kafka Record as headers. But maybe someone has a better idea?

Topic Schemas

Kafka offers the possibility to specify a schema for a topic. Is this out-of-scope because Hono is payload-agnostic? Or could it make sense to support schemas? At least it would have the advantage that Hono would not persist (masses of) invalid messages, which the consumer ignores in the best case. If Ditto is the consumer, this would probably be useful. But the question would be where (and how) to configure it (adapter, tenant, device, ...)?

What about the Well-known Event Message Types? For the Empty Notification, we don't need a schema as it has no payload. For the connection event, which is generated by the protocol adapter, a schema would probably be useful...

Northbound APIs

A Hono instance that uses Kafka instead of AMQP for downstream messages will no longer offer the Telemetry API and the Event API. Do we consider them optional then? Should we in the future specify corresponding "APIs" for Kafka? If so, as a supplement, as an extension of the existing APIs or as a replacement?

tl;dr

• refactoring of the Hono-Client as a first step
• guaranteed message order for the events of a device (not between devices, not for telemetry)
• Topics: event/<tenant-id> and telemetry/<tenant-id>
• Partition keys: for telemetry: null (-> even distribution over partitions, flexible for operation), for events: <device-id> (-> guaranteed message order)
• Change in Hono's behavior: when sending a telemetry message with AT LEAST ONCE delivery, it will no longer know if the message arrived at a consumer

[thjaeckle]

I suggest guaranteeing the order for events, but not for telemetry messages (@thjaeckle, would that work for Ditto?).

I think we would have to take that question to the users of Hono/Ditto combination, whether they require/assume that telemetry messages of a single device sent in high frequency are applied/consumed in order of the sending.

Personally I think this makes sense from a "cost"/ service operations perspective - namely to process "telemetry" messages the lowest possible "cost" (based on Kafka resources and consumers).

I also assume that the default use case for sending telemetry is not that a device sends several telemetry messages in a very short time interval (e.g. multiple within a second).

Can we maybe provide devices to ensure message order when sending QoS 1 telemetry but give no guarantees when sending QoS 0 telemetry?
That way we must not make assumptions but let the device / solution developer decide.

[ctron]

I suggest guaranteeing the order for events, but not for telemetry messages (@thjaeckle, would that work for Ditto?).

I think if you no longer guarantee the order of telemetry, that makes this feature rather useless. Because you never know if the message you just received is already outdated.

[b-abel]

Can we maybe provide devices to ensure message order when sending QoS 1 telemetry but give no guarantees when sending QoS 0 telemetry?

You are right, with QoS 1 telemetry messages are guaranteed to be delivered (i.e. persisted in the case of Kafka) and should therefore be in order.

@ctron I see your point. On the other hand, with QoS 0, if there is no guaranteed delivery, the message might be outdated as well if the update was lost.

Let me give an example where the guaranteed message order could make the problem of outdated messages even worse.
If we think of a device that sends temperature values at a particularly high frequency. Guaranteeing order in Kafka means that all messages from this device must be written to the same partition.
Only one consumer can read from partition, so this prevents parallel message consumption for that device. This could result in an increasing delay of message consumption, where you would always read messages that are outdated in the sense that the values are old.
Without the guaranteed order, you could consume the values with massive parallelism and e.g. display very recent values.
With QoS 0 you could still add a counter or timestamp to filter out outdated messages...

The other aspect is the operating costs. If you have all messages from a device in one partition, and a single device has an exceptionally high data rate, that partition could be larger than others, and you need to provide enough space for that partition on each node in the Kafka cluster.

So I wonder if the guaranteed order for telemetry data with QoS 0 would be worth the cost.

If what I have outlined is a relevant use case, @thjaeckle's proposal allows solution developers to weigh the tradeoff between "fast but unreliable" and "reliable and orderly but slow and expensive" messaging use cases.

Having both options to choose from, depending on the use case, sounds reasonable to me. Otherwise, I feel that telemetry messages and events are too similar to justify the fundamental conceptual distinction between them.

What do you think?

[ctron]

@ctron I see your point. On the other hand, with QoS 0, if there is no guaranteed delivery, the message might be outdated as well if the update was lost.

I think that is not the same. With QoS 0 a message might be dropped and so the current state would be outdated. However, eventually, the next update comes along. Still, all the time to that point, you showed some valid (although outdated) information.

If you mix up the order of delivery, then you might how information that is actually wrong, because the system already received updated information.

Additionally, MQTT with QoS 0 guarantees order of delivery. So we would break that.

[dejanb]

@b-abel @ctron Shouldn't we be able to use "consumer groups" to parallelize consumption from a topic if we want to trade ordering for throughput? So, it should be something that northbound application can choose to do even if we properly partition telemetry topics.

[b-abel]

Additionally, MQTT with QoS 0 guarantees order of delivery. So we would break that.

@ctron You are right. I did not think of that. So, I agree that we should guarantee the message order for telemetry data as well.

@dejanb Sure, a consumer application could always do that anyway.

[sophokles73]

One of the most attractive features of using Kafka for the downstream direction actually is that we can have messages being delivered in order while being able to consume messages in parallel basically at zero cost. I do not see why we shouldn't take advantage of this for telemetry as well. So FMPOV we definitely use multi-partitioned topics for telemetry and events. For telemetry it might make sense to use a (much) higher number of partitions than for events in order to increase the degree of potential parallelism when consuming telemetry.
Apart from that, it should be sufficient to keep message order per device, thus using the tenant/device-id in the partitioning key seems appropriate to me.

[sophokles73]

When creating a tenant the topics event/ and telemetry/ should be created (+ Kafka account + ACLs).

+1

[sophokles73]

At least for telemetry messages with AT LEAST ONCE delivery, the device currently knows immediately whether a consumer is connected. In this respect, Hono's behavior changes from the point of view of the device. This is probably inevitable.

I agree. We will need to bite that bullet for the Kafka variant. On the other hand, we are not very consistent with providing this application level end-to-end ACK mechanism anyway. In fact, we only support it for QoS 1 telemetry. We might want to think about a way to introduce generic application level end-to-end ACKing as part of #1276 ...

[sophokles73]

Regarding the message format, it would probably be most similar to the current behavior if the message just contains the payload, and all (application) properties Hono sets now are added to the Kafka Record as headers. But maybe someone has a better idea?

Is there a potential problem with that approach? Does it have any particular drawbacks?

[sophokles73]

A Hono instance that uses Kafka instead of AMQP for downstream messages will no longer offer the Telemetry API and the Event API. Do we consider them optional then? Should we in the future specify corresponding "APIs" for Kafka? If so, as a supplement, as an extension of the existing APIs or as a replacement?

FMPOV we should simply define Kafka variants of the APIs, i.e. they will be defined by means of Kafka topics and the Kafka protocol/client. An operator/administrator can then choose to either provide the AMQP 1.0 or Kafka-based north bound APIs ...

[sophokles73]

Kafka offers the possibility to specify a schema for a topic. Is this out-of-scope because Hono is payload-agnostic? Or could it make sense to support schemas?

Can you provide a pointer to the corresponding Kafka documentation? Based on a gut feeling, I do not believe that this would be very helpful. We also do not use the implicit schema support of AMQP 1.0, so I do not see why we would use it here ...

[b-abel]

Kafka offers the possibility to specify a schema for a topic. Is this out-of-scope because Hono is payload-agnostic? Or could it make sense to support schemas?

Can you provide a pointer to the corresponding Kafka documentation? Based on a gut feeling, I do not believe that this would be very helpful. We also do not use the implicit schema support of AMQP 1.0, so I do not see why we would use it here ...

I was not quite correct in the terminology. Sorry about that. The term "schema" is not used by Kafka, but only by external extensions (e.g. Confluent Schema Registry). At the Kafka client, you just configure serializers and deserializers (which in turn can use schema registries). For Hono it is only a question of whether one can configure own serializers at the producer (and if so, on which level, per protocol adapter instance, per tenant, or per device?). This is a special case of the support of custom producer configurations, which is still to be clarified.

[dejanb]

Kafka offers the possibility to specify a schema for a topic. Is this out-of-scope because Hono is payload-agnostic? Or could it make sense to support schemas?

Can you provide a pointer to the corresponding Kafka documentation? Based on a gut feeling, I do not believe that this would be very helpful. We also do not use the implicit schema support of AMQP 1.0, so I do not see why we would use it here ...

I was not quite correct in the terminology. Sorry about that. The term "schema" is not used by Kafka, but only by external extensions (e.g. Confluent Schema Registry). At the Kafka client, you just configure serializers and deserializers (which in turn can use schema registries). For Hono it is only a question of whether one can configure own serializers at the producer (and if so, on which level, per protocol adapter instance, per tenant, or per device?). This is a special case of the support of custom producer configurations, which is still to be clarified.

I would leave Hono content-agnostic and allow producers and consumers on the south and north bounds define schemas if necessary.

[b-abel]

The first step is to replace AMQP with Kafka for the downstream direction. Commands are out of scope for now.
For this, we need a Hono Client that uses Kafka instead of AMQP for (downstream) messaging.
As a prerequisite, we should refactor the Hono Client so that it no longer exposes AMQP specifics to the outside (i.e. to the protocol adapter) and can be used protocol-independently with AMQP and Kafka.

I created #2156 for that.

[b-abel]

Regarding the message format, it would probably be most similar to the current behavior if the message just contains the payload, and all (application) properties Hono sets now are added to the Kafka Record as headers. But maybe someone has a better idea?

Is there a potential problem with that approach? Does it have any particular drawbacks?

AMQP 1.0 defines different sections of a message that contain different types of metadata. In Hono we use at least message properties (creation-time), headers (time-to-live, durable), application properties (device_id, orig_address, orig_adapter, ttd, ...).
Kafka only knows headers. So, all metadata must be put there (we decided against the alternative of defining a complex message that contains payload and metadata).

If we map the different namespaces of metadata from AMQP messages to the flat structure of headers, we have to handle name conflicts. I don't think there will be "legitimate" conflicts, but devices could change values without being authorized by putting the same key in a different section (potentially security-relevant?). At least with the AMQP adapter, the device can set arbitrary application properties and thus potentially overwrite values.

We should therefore define an order in which they overwrite each other and make sure that all relevant properties are always set. (Alternatively, we could prefix the headers to indicate which AMQP message section they come from. However, I would rather avoid adding AMQP specifics to our Kafka APIs.)

To be precise, as with HTTP, there can of course be multiple headers with the same key on a message, but I don't see that we should use that at the moment. The new client interface (#2156) provides a Map<String, ?> for the metadata, but I would like to use only String (maybe also integers and booleans) for the values.

Regarding the implementation I see two possibilities: Either our Kafka-based client defines headers for the important metadata and sets and validates them. Or we can keep the client dumb and simply add everything passed to it in the properties parameter as headers. The latter would mean that we could continue to use the logic implemented mainly in MessageHelper and the protocol adapters.

WDYT?

[b-abel]

Additionally, we have yet to analyze where Hono is using AMQP capabilities to provide features that we need to define and implement for Kafka-based messaging. For example: Do we want to define a time-to-live header in the Kafka-based APIs that the consumer has to consider?

[sophokles73]

Regarding the implementation I see two possibilities: Either our Kafka-based client defines headers for the important metadata and sets and validates them. Or we can keep the client dumb and simply add everything passed to it in the properties parameter as headers. The latter would mean that we could continue to use the logic implemented mainly in MessageHelper and the protocol adapters.

This is the approach that I am currently following for the AMQP 1.0 based implementation of the adapter client.

Additionally, we have yet to analyze where Hono is using AMQP capabilities to provide features that we need to define and implement for Kafka-based messaging. For example: Do we want to define a time-to-live header in the Kafka-based APIs that the consumer has to consider?

Passing on a device provided TTL makes sense FMPOV as the device should know best for how long the information should be considered relevant/accurate. We should explicitly define such well known headers in the Kafka based API specs.

[b-abel]

We have decided that event messages will be sorted per device and telemetry messages as well. We also agreed to use separate topics for events and telemetry messages (for each client).

I just noticed that we did not explicitly name the consequence here: Events and telemetry messages of the same device are sorted separately. Backend applications, therefore, have to process both types of messages independently (as long as the order is important).
@ctron @dejanb @sophokles73 Are you aware of this? And do you agree with it?

My take on this:

Even if some users might spontaneously consider a comprehensive order between events and telemetry messages (per device) desirable, I think we should stick to the separate topics.

With AMQP the main difference between events and telemetry messages is that the former are persisted. With Kafka as a messaging system, this applies to all messages. If events and telemetry messages (with QoS1) were stored in the same topic and differed only by their headers, I would not find the conceptual distinction justified. But I assume that we want to keep the two message types and so I wonder how to put the difference in a nutshell.

I understand that we expect a lot of telemetry messages, events should be sent orders of magnitude less frequently, but have higher importance. Separate topics have the advantage that the storage times for events and telemetry messages can be configured separately. Longer storage can be configured to achieve higher reliability for events. On the other hand, events can be used as a kind of "priority queue" if they are stored and read separately. Since Kafka does not offer the ability to consume messages of a partition pre-filtered, the backend application may have to consume thousands of telemetry messages under load until it can read the one important event message.

This way, _separate topics enable Hono to make events faster and more durable than the cheaper telemetry messages_. Of course, the differences and consequences must be clearly stated in the documentation.

[sophokles73]

Backend applications, therefore, have to process both types of messages independently (as long as the order is important).

I do not see any difference to the way it works with AMQP 1.0. We have two separate and independent endpoints there as well ...

[iamrsaravana]

We could try to use this AMQP Kafka bridge (https://github.com/rhiot/amqp-kafka-bridge) for this issue.

Could you share the steps how to send hono data into Strimzi Kafka

[iamrsaravana]

Has anybody tried AMQP camel source connector for t to read data from Hono and send to Kafka? . Could you share your thoughts why we cann't use it.

[sophokles73]

Has anybody tried AMQP camel source connector for t to read data from Hono and send to Kafka?

At least I haven't.

Could you share your thoughts why we cann't use it.

I do not see why you shouldn't be able to use it. However, you would then still need an AMQP 1.0 Messaging Network in place. The overall idea here is to support Kafka as an alternative to the AMQP 1.0 Messaging Network.

[iamrsaravana]

Yes, agreed still we need AMQP 1.0 Messaging network, after that we can push data directly to Kafka source connector. Hope this would avoid converting from AMQP 1.0 messages to a Kafka producer.

When can we expect Kafka protocol binding in Hono(https://www.eclipse.org/ditto/connectivity-protocol-bindings-kafka2.html)

[sophokles73]

This issue is scheduled for the 1.6.0 release which we hope to do end of January 2021

[iamrsaravana]

This issue is scheduled for the 1.6.0 release which we hope to do end of January 2021

Thanks for confirmation.

[iamrsaravana]

Now I am able to send the data from Hono to Strimzi Kafka Connector using Camel AMQP Source Connector.

Follow the below steps to pull data from Hono and send to Strimzi Kafka Connector:

1) Creating Name space:
kubectl create namespace kafka

2) install latest Strimzi operator using the below command:

kubectl apply -f 'https://strimzi.io/install/latest?namespace=kafka' -n kafka
3) creating Kafka (1 pod) and zookeeper(2 pods ) using the below yaml configuration:

    **kafka.yaml:**

    apiVersion: kafka.strimzi.io/v1beta1
    kind: Kafka
    metadata:
      name: my-cluster
    spec:
      kafka:
        version: 2.6.0
        replicas: 1
        listeners:
          - name: plain
            port: 9092
            type: internal
            tls: false
          - name: tls
            port: 9093
            type: internal
            tls: true
        config:
          offsets.topic.replication.factor: 1
          transaction.state.log.replication.factor: 1
          transaction.state.log.min.isr: 1
          log.message.format.version: "2.6"
        storage:
          type: jbod
          volumes:
          - id: 0
            type: persistent-claim
            size: 10Gi
            deleteClaim: false
      zookeeper:
        replicas: 2
        storage:
          type: persistent-claim
          size: 10Gi    
          deleteClaim: false
      entityOperator:
        topicOperator: {}
        userOperator: {}

    Kuebctl apply -f kafka.yaml  -n kafka.

    _ps: Here I use glusterfs persistent storage._

4) Creating AMQP - Source Connector image file using below commands:

I downloaded Camel-amqp-kafka-connector, JMS Jar files from below link:

        https://repo.maven.apache.org/maven2/org/apache/camel/kafkaconnector/camel-amqp-kafka-connector/0.7.0/camel-amqp-kafka-connector-0.7.0-package.tar.gz
        https://downloads.apache.org/qpid/jms/0.51.0/apache-qpid-jms-0.51.0-bin.tar.gz

        After downloaded the above tar and unzipped and created docker image file using below command

        **Dockerfile:**
        FROM strimzi/kafka:0.20.1-kafka-2.6.0
        USER root:root
        RUN mkdir -p /opt/kafka/plugins/camel-kafka-connectors/camel-amqp-kafka-connector/
        COPY ./plugin/apache-qpid-jms-0.55.0/* /opt/kafka/plugins/camel-kafka-connectors/camel-amqp-kafka-connector/
        COPY ./camel-amqp-kafka-connector/* /opt/kafka/plugins/camel-kafka-connectors/camel-amqp-kafka-connector/
        USER 1001


        Docker build -f ./Dockerfile -t localhost:5000/my-connector-amqp_new .
        docker push localhost:5000/my-connector-amqp_new

       _ps: Here I have used local docker repository_

5) Creating Kafkaconnect using below Yaml configuration:

    Kafkaconnect.yaml:

    apiVersion: kafka.strimzi.io/v1beta1
    kind: KafkaConnect
    metadata:
      name: my-connect-cluster-new
      annotations:
        strimzi.io/use-connector-resources: "true"
    spec:
      image: 10.128.0.6:5000/my-connector-amqp_new5
      replicas: 1
      bootstrapServers: my-cluster-kafka-bootstrap:9092
      config:
        group.id: connect-cluster-new
        offset.storage.topic: connect-cluster-offsets
        config.storage.topic: connect-cluster-configs
        status.storage.topic: connect-cluster-status
        config.storage.replication.factor: 1
        offset.storage.replication.factor: 1
        status.storage.replication.factor: 1

  kubectl apply -f Kafkaconnect.yaml -n kafka

6) Creating AMQP-KafkaConnector using below yaml file:

amqp_connector.yaml

    apiVersion: kafka.strimzi.io/v1alpha1
    kind: KafkaConnector
    metadata:
      name: camelamqpsourceconnector
      labels:
        strimzi.io/cluster: my-connect-cluster-new
    spec:
      class: org.apache.camel.kafkaconnector.amqp.CamelAmqpSourceConnector
      tasksMax: 1
      config:
        camel.component.amqp.includeAmqpAnnotations: true
        camel.component.amqp.connectionFactory: '#class:org.apache.qpid.jms.JmsConnectionFactory'
        camel.component.amqp.connectionFactory.remoteURI: amqp://10.106.88.243:15672
        camel.component.amqp.username: consumer@HONO
        camel.component.amqp.password: verysecret
        camel.component.amqp.testConnectionOnStartup: true
        camel.source.path.destinationType: queue 
        #( my usecase i am reading from hono)
        camel.source.path.destinationName: telemetry/LTTS_TENANT
        # ( this is the place where hono writing the data)
        topics: mytopic 
        #(kafka topics where you want to write it)

kubectl apply -f amqp_connector.yaml -n kafka.
ps: IP address given here all are my pod ip address. it may vary according to your pod.

7 ) viewing data in strimzi kafka using the below command:
kubectl -n kafka exec my-cluster-kafka-0 -c kafka -i -t -- bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic mytopic

I Hope this may be useful for somebody who wants to pull the data directly using AMQP Camel Connector.

[gaochundong]

thanks for the discussion.

[andrew-kratik]

Hi @sophokles73
Thanks for your great work!
Can you clarify when Kafka support (and Hono 1.7.0) is expected to be released?

[sophokles73]

We are planning to release 1.7.0 this week. Note that support for Kafka is still considered experimental, though.