New: Cybus Helm Charts 3.0.0.
Learn more.
Ctrlk
Contact Uscybus.io
For the complete documentation index, see llms.txt. This page is also available as Markdown.

Apache Kafka Integration

How to integrate a self-managed Apache Kafka cluster with Connectware, including PLAINTEXT, SASL SCRAM, and mutual TLS security, producing shop floor data to a topic, and consuming a topic into MQTT.

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 with sufficient user permissions.

  • Basic knowledge of MQTT and the Connectware services concept (for example, service commissioning files, connections, and endpoints).

Connectware and Apache Kafka Integration

Connectware communicates with Apache Kafka through the Kafka connector. 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 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 guide, which authenticates with a connection string.

The MQTT topics in this guide follow an ISA-95-style equipment hierarchy (<enterprise>/<site>/<area>/<line>/<cell>). 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.

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.

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.

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.

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.

Create the SCRAM credentials for the Connectware client on your cluster:

Replace ${SASL_PASSWORD} with the password for the Connectware client. For more information, see Authentication using SASL/SCRAM in the Apache Kafka documentation.

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 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 in the Apache Kafka documentation.

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). 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.

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:

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.

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), so the transform rule extracts it before the mapping publishes it to MQTT.

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.

  5. The result of every produce request is also published to the /res topic of the endpoint. Use the Data Explorer 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

apache-kafka-example.yml
4KB
Open
PreviousMessaging & Event StreamingNextConfluent Cloud Integration

Last updated

Was this helpful?