> For the complete documentation index, see [llms.txt](https://docs.cybus.io/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.cybus.io/2-4-2/guides/system-connectivity/messaging-event-streaming/apache-kafka-integration.md). # Apache Kafka Integration This guide describes how to integrate a self-managed Apache Kafka cluster with Connectware. You configure a service commissioning file that streams shop floor data to a Kafka topic through the Kafka connector and consumes a Kafka topic into the MQTT topic hierarchy in return. A complete example file is available at the end of this guide. ## Objectives * Connecting Connectware to an Apache Kafka cluster, from an unauthenticated development setup to SASL SCRAM and mutual TLS. * Creating the Kafka topics for both directions. * Producing shop floor data from the MQTT topic hierarchy to a Kafka topic. * Consuming a Kafka topic into an MQTT topic. ## Prerequisites To follow this guide, you will need the following: * A running instance of Cybus Connectware. * A self-managed Apache Kafka cluster and network access from Connectware to the brokers. Kafka clients connect to every broker directly, so Connectware must be able to reach each address that the brokers advertise in their `advertised.listeners` configuration, not only the bootstrap address. * Administrative access to the cluster to create topics and, depending on the security configuration, SCRAM credentials or client certificates. * Access to the [Admin UI](/2-4-2/access/admin-ui.md) with sufficient [user permissions](/2-4-2/access/user-management.md). * Basic knowledge of MQTT and the Connectware [services](/2-4-2/data-flows/services.md) concept (for example, [service commissioning files](/2-4-2/data-flows/service-commissioning-files.md), [connections](/2-4-2/data-flows/service-commissioning-files/resources/cybus-connection.md), and [endpoints](/2-4-2/data-flows/service-commissioning-files/resources/cybus-endpoint.md)). ## Connectware and Apache Kafka Integration Connectware communicates with Apache Kafka through the [Kafka connector](/2-4-2/connectors/enterprise-connectors/kafka.md). Two connector characteristics shape the setup: * **A connection is either a producer or a consumer.** The `clientType` property of the connection decides whether its endpoints write to Kafka or subscribe to Kafka. For the bidirectional integration in this guide, you define two connections to the same cluster. * **The connector produces plain message values.** Connectware sends the message payload as a string, typically serialized JSON. Serialization formats that require a schema registry, such as Avro or Protobuf, are not part of the Kafka connector and are out of scope for this guide. The security configuration of the connection must match the listener that your brokers expose. Without the `sasl` and `mutualAuthentication` properties, the connection is unencrypted and unauthenticated (a `PLAINTEXT` listener). Configuring the `sasl` property automatically enables TLS, so the matching listener on the broker side is `SASL_SSL`. Configuring the `mutualAuthentication` property enables TLS with client certificates (an `SSL` listener). All three variants are covered in this guide. This guide covers the general case of a self-managed cluster. If you connect to Confluent Cloud, see the [Confluent Cloud Integration](/2-4-2/guides/system-connectivity/messaging-event-streaming/confluent-cloud-integration.md) guide, which uses cluster API keys with SASL PLAIN. If you connect to the Kafka endpoint of Azure Event Hubs, see the [Azure Event Hubs Integration](/2-4-2/guides/system-connectivity/messaging-event-streaming/azure-event-hubs-integration.md) guide, which authenticates with a connection string. The MQTT topics in this guide follow an ISA-95-style equipment hierarchy (`////`). The producer mapping subscribes with wildcards across all levels, so any machine in the hierarchy is picked up without changing the integration. ### Kafka Connection Properties The connection to Apache Kafka requires the broker address and, for the recommended production setup, SASL credentials. We add them as parameters to the service commissioning file, so you can set them when you install the service. Do not worry about copying the service commissioning file snippets together into one, the complete example file is available at the end of this guide. * `brokers`: The address of a broker of your cluster, including the port. The `brokers` property of the connection is an array, and the client discovers the remaining brokers from the first one it reaches. Listing more than one broker makes the initial connection resilient against a single broker being down. This guide uses port `9092` for the `PLAINTEXT` listener and port `9093` for the TLS listeners, matching the common convention. Use the ports of your cluster's listener configuration. * `saslUsername` and `saslPassword`: The SCRAM credentials of the Connectware client. * `produceTopic`: The Kafka topic that receives the shop floor data. * `consumeTopic`: The Kafka topic that Connectware consumes into MQTT. * `consumerGroupId`: The consumer group ID that Connectware uses when consuming. If you do not set the `groupId` property, it defaults to the ID of the connection resource. * `topicRoot`: The root of the MQTT topic hierarchy. Defaults to `enterprise`. {% code lineNumbers="true" %} ```yaml parameters: brokers: description: Broker address of your Apache Kafka cluster type: string default: kafka-1.example.com:9093 saslUsername: description: SASL username of the Connectware client type: string default: connectware saslPassword: description: SASL password of the Connectware client type: string produceTopic: description: Kafka topic that receives the shop floor data type: string default: factory.machine-data consumeTopic: description: Kafka topic that Connectware consumes into MQTT type: string default: factory.commands consumerGroupId: description: Consumer group ID that Connectware uses when consuming type: string default: connectware topicRoot: description: Root of the MQTT topic hierarchy type: string default: enterprise ``` {% endcode %} ### Creating the Kafka Topics Create the topics for both directions on your cluster before you install the service. Apache Kafka creates topics automatically by default when a client uses them, but auto-created topics get the broker default partition count and replication factor, which is rarely what you want in production. {% code lineNumbers="true" %} ```bash bin/kafka-topics.sh --create \ --bootstrap-server kafka-1.example.com:9092 \ --topic factory.machine-data \ --partitions 6 \ --replication-factor 3 bin/kafka-topics.sh --create \ --bootstrap-server kafka-1.example.com:9092 \ --topic factory.commands \ --partitions 6 \ --replication-factor 3 ``` {% endcode %} Adjust the partition count and replication factor to your cluster. On a single-broker development cluster, use a replication factor of one. ### Choosing the Security Configuration The following sections show the producer connection in three security configurations. Pick the one that matches your cluster listener. The consumer connection uses the same security properties. #### Development Setup Without Authentication For a local development cluster with a `PLAINTEXT` listener, the connection only needs the broker address. The `clientType` property defaults to `producer`. {% code lineNumbers="true" %} ```yaml resources: kafkaProducerConnection: type: Cybus::Connection properties: protocol: Kafka connection: brokers: - kafka-1.example.com:9092 ``` {% endcode %} {% hint style="warning" %} A `PLAINTEXT` connection is unencrypted and unauthenticated. Anyone with network access to the broker can read and write all topics. Use this configuration only for local development, never in production. {% endhint %} #### SASL SCRAM Authentication over TLS For production, we recommend SASL with the `scram-sha-512` mechanism. Unlike SASL PLAIN, SCRAM does not store the clear text password on the broker, and the challenge-response handshake never transmits it. Configuring the `sasl` property makes the connector connect over TLS, so the broker listener must be `SASL_SSL`. For all available properties, see [Kafka Connection Properties](/2-4-2/connectors/enterprise-connectors/kafka/kafkaconnection.md). Create the SCRAM credentials for the Connectware client on your cluster: {% code lineNumbers="true" %} ```bash bin/kafka-configs.sh --bootstrap-server kafka-1.example.com:9092 \ --alter \ --add-config 'SCRAM-SHA-512=[password=${SASL_PASSWORD}]' \ --entity-type users \ --entity-name connectware ``` {% endcode %} Replace `${SASL_PASSWORD}` with the password for the Connectware client. For more information, see [Authentication using SASL/SCRAM](https://kafka.apache.org/documentation/#security_sasl_scram) in the Apache Kafka documentation. {% code lineNumbers="true" %} ```yaml kafkaProducerConnection: type: Cybus::Connection properties: protocol: Kafka connection: brokers: - !ref brokers clientType: producer sasl: mechanism: scram-sha-512 username: !ref saslUsername password: !ref saslPassword ``` {% endcode %} If your brokers use TLS certificates issued by an internal certificate authority (CA), add the `caCert` property with the Base64-encoded content of the CA certificate PEM file, for example encoded with `base64 -w 0 ca.crt` on Linux or `base64 -i ca.crt` on macOS. Alternatively, `trustAllCertificates: true` disables certificate validation entirely. This makes the connection vulnerable to machine-in-the-middle attacks, so use it only for testing. If your cluster enforces access control lists (ACLs), the Connectware principal needs write access to the producer topic, read access to the consumer topic, and read access to the consumer group. For more information, see [Authorization and ACLs](https://kafka.apache.org/documentation/#security_authz) in the Apache Kafka documentation. #### Mutual TLS Authentication If your cluster authenticates clients with certificates instead of SASL credentials, set the `mutualAuthentication` property to `true` and provide the client certificate, the client private key, and the CA certificate. All three properties take the Base64-encoded content of the respective PEM file. The broker listener for this configuration is `SSL`. For more information, see [Encryption and Authentication using SSL](https://kafka.apache.org/documentation/#security_ssl) in the Apache Kafka documentation. {% code lineNumbers="true" %} ```yaml kafkaProducerConnection: type: Cybus::Connection properties: protocol: Kafka connection: brokers: - kafka-1.example.com:9093 clientType: producer mutualAuthentication: true caCert: !ref caCert clientCert: !ref clientCert clientPrivateKey: !ref clientPrivateKey ``` {% endcode %} The remainder of this guide and the complete example file use the SASL SCRAM connection. ### Producing Shop Floor Data to Kafka A write endpoint addresses the Kafka topic, and a mapping feeds it from the MQTT topic hierarchy. The Kafka connector expects a JSON message with the payload as `value` (see [Message assembly](/2-4-2/connectors/enterprise-connectors/kafka.md#message-assembly)). The `transform` rule wraps the incoming payload accordingly and serializes it to a string with `$string`, so machines can publish their data without knowing about this convention. {% code lineNumbers="true" %} ```yaml machineDataEndpoint: type: Cybus::Endpoint properties: protocol: Kafka connection: !ref kafkaProducerConnection write: topic: !ref produceTopic machineDataMapping: type: Cybus::Mapping properties: mappings: - subscribe: topic: !sub '${topicRoot}/+site/+area/+line/+cell/machine-data' publish: endpoint: !ref machineDataEndpoint rules: - transform: expression: '{ "value": [ { "value": $string($) } ] }' ``` {% endcode %} Any message published to a matching topic, for example `enterprise/hamburg/assembly/line-1/press-01/machine-data`, is now produced to the Kafka topic as one record: {% code lineNumbers="true" %} ```json { "temperature": 42.1, "pressure": 5.4, "serial": "SN-0001" } ``` {% endcode %} Each record in the `value` array also accepts optional `key`, `partition`, `headers`, and `timestamp` fields. Set a `key` if you want Kafka to keep all records of one machine on the same partition, which preserves their order for consumers. The message can also override the endpoint `topic` and `acks` settings per request. By default, the endpoint waits until all in-sync replicas have acknowledged a record (`acks: -1`), which is the safest setting. For all available fields, see [Kafka Endpoint Properties](/2-4-2/connectors/enterprise-connectors/kafka/kafkaendpoint.md). ### Consuming a Kafka Topic into MQTT For the opposite direction, for example to receive commands or analytics results from other Kafka clients, we set up a second connection with the `clientType` property set to `consumer`. A connection can only take all write endpoints or all subscribe endpoints, but not both at the same time, which is why the producer connection cannot be reused. The subscribe endpoint consumes the Kafka topic. With `fromBeginning: false`, the consumer starts at the end of the topic and only receives new records. The consumed message contains the record body in the `value.message` field (see [Message assembly](/2-4-2/connectors/enterprise-connectors/kafka.md#message-assembly)), so the `transform` rule extracts it before the mapping publishes it to MQTT. {% code lineNumbers="true" %} ```yaml kafkaConsumerConnection: type: Cybus::Connection properties: protocol: Kafka connection: brokers: - !ref brokers clientType: consumer groupId: !ref consumerGroupId sasl: mechanism: scram-sha-512 username: !ref saslUsername password: !ref saslPassword commandsEndpoint: type: Cybus::Endpoint properties: protocol: Kafka connection: !ref kafkaConsumerConnection subscribe: topic: !ref consumeTopic fromBeginning: false commandsMapping: type: Cybus::Mapping properties: mappings: - subscribe: endpoint: !ref commandsEndpoint publish: topic: !sub '${topicRoot}/kafka/commands' rules: - transform: expression: 'value.message' ``` {% endcode %} Every record that another client produces to the consume topic now appears on the `enterprise/kafka/commands` MQTT topic, where any other Connectware service can pick it up, for example to write a setpoint to a PLC. ## Verifying the Integration 1. Install the service and set the parameters with the values of your cluster. 2. Check that both connections are in the **Connected** state on the service details page in the Admin UI. If the credentials are wrong or the security configuration does not match the broker listener, the connections do not reach the connected state. 3. Publish a test message to `enterprise/hamburg/assembly/line-1/press-01/machine-data`, for example with an MQTT client or the Admin UI. 4. Read the produced record from the topic with the console consumer of your Kafka installation. On a secured cluster, pass a client configuration file with the `--consumer.config` option. 
bin/kafka-console-consumer.sh \
     --bootstrap-server kafka-1.example.com:9092 \
     --topic factory.machine-data \
     --from-beginning
5. The result of every produce request is also published to the `/res` topic of the endpoint. Use the [Data Explorer](/2-4-2/monitoring/data-explorer.md) to inspect it. On success, the result contains `errorCode: 0` and the offset of the produced record. If the broker rejects a request, the message on the `/res` topic contains an `error` property instead. 6. Produce a test record to the consume topic, for example with the `kafka-console-producer.sh` tool, and check that it appears on the `enterprise/kafka/commands` MQTT topic in the Data Explorer. ## Service Commissioning File Example {% file src="/files/P7FKXTNqnjq1x75noYKq" %} {% code title="apache-kafka-example.yml" lineNumbers="true" expandable="true" %} ```yaml --- # ---------------------------------------------------------------------------- # Service Commissioning File # ---------------------------------------------------------------------------- # Copyright: Cybus GmbH # Contact: support@cybus.io # ---------------------------------------------------------------------------- # Apache Kafka Integration (Example) # ---------------------------------------------------------------------------- description: > Service commissioning file for the bidirectional integration between Connectware and a self-managed Apache Kafka cluster (Example) metadata: name: Apache Kafka Integration provider: cybus homepage: https://www.cybus.io version: 1.0.0 parameters: brokers: description: Broker address of your Apache Kafka cluster type: string default: kafka-1.example.com:9093 saslUsername: description: SASL username of the Connectware client type: string default: connectware saslPassword: description: SASL password of the Connectware client type: string produceTopic: description: Kafka topic that receives the shop floor data type: string default: factory.machine-data consumeTopic: description: Kafka topic that Connectware consumes into MQTT type: string default: factory.commands consumerGroupId: description: Consumer group ID that Connectware uses when consuming type: string default: connectware topicRoot: description: Root of the MQTT topic hierarchy type: string default: enterprise resources: # Producer connection using SASL SCRAM-SHA-512. # Configuring the sasl property makes the connector connect over TLS. # If your brokers use certificates from an internal CA, add the caCert # property with the Base64-encoded PEM content of the CA certificate. kafkaProducerConnection: type: Cybus::Connection properties: protocol: Kafka connection: brokers: - !ref brokers clientType: producer sasl: mechanism: scram-sha-512 username: !ref saslUsername password: !ref saslPassword # Produces shop floor data to the Kafka topic machineDataEndpoint: type: Cybus::Endpoint properties: protocol: Kafka connection: !ref kafkaProducerConnection write: topic: !ref produceTopic # Streams shop floor data from the MQTT topic hierarchy to Kafka. # The transform rule wraps each payload in the message format that the # Kafka connector expects. machineDataMapping: type: Cybus::Mapping properties: mappings: - subscribe: topic: !sub '${topicRoot}/+site/+area/+line/+cell/machine-data' publish: endpoint: !ref machineDataEndpoint rules: - transform: expression: '{ "value": [ { "value": $string($) } ] }' # Consumer connection to the same cluster. # A Kafka connection is either a producer or a consumer, never both. kafkaConsumerConnection: type: Cybus::Connection properties: protocol: Kafka connection: brokers: - !ref brokers clientType: consumer groupId: !ref consumerGroupId sasl: mechanism: scram-sha-512 username: !ref saslUsername password: !ref saslPassword # Consumes new records from the Kafka topic commandsEndpoint: type: Cybus::Endpoint properties: protocol: Kafka connection: !ref kafkaConsumerConnection subscribe: topic: !ref consumeTopic fromBeginning: false # Publishes the body of every consumed record to an MQTT topic commandsMapping: type: Cybus::Mapping properties: mappings: - subscribe: endpoint: !ref commandsEndpoint publish: topic: !sub '${topicRoot}/kafka/commands' rules: - transform: expression: 'value.message' ``` {% endcode %} --- # Agent Instructions This documentation is published with GitBook. GitBook is the documentation platform designed so that both humans and AI agents can read, navigate, and reason over technical content effectively. Learn more at gitbook.com. ## Querying This Documentation If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question. Perform an HTTP GET request on the current page URL with the `ask` query parameter, and the optional `goal` query parameter: ``` GET https://docs.cybus.io/2-4-2/guides/system-connectivity/messaging-event-streaming/apache-kafka-integration.md?ask=&goal= ``` `ask` is the immediate question: it should be specific, self-contained, and written in natural language. `goal` is optional and describes the broader end goal you are ultimately trying to accomplish on behalf of the user. GitBook uses it to tailor the answer towards what is most useful for that goal. The response will contain a direct answer to the question and relevant excerpts and sources from the documentation. Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.