> 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/redpanda-integration.md). # Redpanda Integration This guide describes how to integrate Redpanda with Connectware. You configure a service commissioning file that streams shop floor data to a Redpanda topic through the Kafka connector and consumes a Redpanda topic into the MQTT topic hierarchy in return. The setup works for self-hosted Redpanda clusters and for Redpanda Cloud. A complete example file is available at the end of this guide. ## Objectives * Connecting Connectware to a Redpanda cluster through the Kafka connector. * Producing shop floor data from the MQTT topic hierarchy to a Redpanda topic. * Consuming a Redpanda topic into an MQTT topic. * Securing the connection with SASL/SCRAM on self-hosted clusters and Redpanda Cloud. ## Prerequisites To follow this guide, you will need the following: * A running instance of Cybus Connectware. * A Redpanda cluster. Either a self-hosted cluster whose Kafka API is reachable from Connectware (port `9092` by default) and the `rpk` command line client, or a Redpanda Cloud cluster. * 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 Redpanda Integration Redpanda is a streaming platform with a Kafka-compatible API. Clients built for Apache Kafka 0.11 or later work with Redpanda without changes (see [Kafka Compatibility](https://docs.redpanda.com/redpanda-cloud/develop/kafka-clients/)). Connectware communicates with Redpanda through the [Kafka connector](/2-4-2/connectors/enterprise-connectors/kafka.md), with no Redpanda-specific configuration. If you use Confluent Cloud instead, see [Confluent Cloud Integration](/2-4-2/guides/system-connectivity/messaging-event-streaming/confluent-cloud-integration.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 Redpanda or subscribe to Redpanda. 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. Avro or Protobuf serialization with a schema registry is not part of the Kafka connector and is out of scope for this guide. 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. ### Preparing a Self-Hosted Cluster Redpanda disables automatic topic creation by default (the `auto_create_topics_enabled` cluster property), so create the topics before you install the service. On any host with access to the cluster, use the [rpk topic create](https://docs.redpanda.com/current/reference/rpk/rpk-topic/rpk-topic-create/) command: {% code lineNumbers="true" %} ```bash rpk topic create factory.machine-data factory.commands ``` {% endcode %} Use the `-p` flag to set the number of partitions and the `-r` flag to set the replication factor. Without these flags, the cluster defaults apply. For a quick test on an unsecured development cluster, this is all the preparation you need and you can skip ahead to [Redpanda Connection Properties](#redpanda-connection-properties). For production, enable SASL/SCRAM authentication. Create a superuser, add it to the `superusers` cluster property, and enable SASL on the Kafka API: {% code lineNumbers="true" %} ```bash rpk security user create admin -p '${ADMIN_PASSWORD}' --mechanism SCRAM-SHA-256 rpk cluster config set superusers '["admin"]' rpk cluster config set enable_sasl true ``` {% endcode %} * `${ADMIN_PASSWORD}`: The password of the superuser. Then create a user for Connectware and grant it access control lists (ACLs) for the two topics and the consumer group. Once SASL is enabled, `rpk` commands must authenticate, for example with the `-X user`, `-X pass`, and `-X sasl.mechanism` flags of the superuser. {% code lineNumbers="true" %} ```bash rpk security user create connectware -p '${SASL_PASSWORD}' --mechanism SCRAM-SHA-256 rpk security acl create --allow-principal User:connectware --operation write --topic factory.machine-data rpk security acl create --allow-principal User:connectware --operation read --topic factory.commands rpk security acl create --allow-principal User:connectware --operation read --group connectware ``` {% endcode %} * `${SASL_PASSWORD}`: The password of the Connectware user. You set it as a parameter when you install the service. Redpanda supports the `SCRAM-SHA-256` and `SCRAM-SHA-512` mechanisms. Instead of enabling SASL globally, you can also enable authentication per listener with the `kafka_enable_authorization` cluster property and the `authentication_method` listener property. For more information, see [Configure Authentication](https://docs.redpanda.com/current/manage/security/authentication/). {% hint style="warning" %} Configuring the `sasl` property on a Connectware Kafka connection automatically enables TLS. The Kafka API listener of your Redpanda cluster must therefore serve TLS (see [Configure Kafka TLS Encryption](https://docs.redpanda.com/current/manage/security/encryption/)). If the listener uses a self-signed certificate, set the `caCert` property of the connection to the Base64 encoded PEM content of the root CA certificate, or set `trustAllCertificates: true` for testing. {% endhint %} ### Preparing a Redpanda Cloud Cluster For Redpanda Cloud, the cluster is already secured. TLS is always enabled and clients authenticate with SASL (see [Cloud Authentication](https://docs.redpanda.com/cloud-data-platform/security/cloud-authentication/)). You need the following from the Redpanda Cloud UI: * **The bootstrap server URL.** You find it on the **Overview** page of your cluster, in the **Kafka API** section. * **A user with SCRAM credentials.** Create the user on the **Security** page of your cluster and select the `SCRAM-SHA-256` or `SCRAM-SHA-512` mechanism. Grant the user ACLs with write access to the producer topic, read access to the consumer topic, and read access to the consumer group. * **The topics.** Create them on the **Topics** page or with the `rpk topic create` command. Redpanda Cloud broker certificates are signed by a public certificate authority, so you do not need to configure the `caCert` or `trustAllCertificates` properties. ### Redpanda Connection Properties The connection to Redpanda requires the bootstrap server address and, for a secured cluster, the 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. * `bootstrapServer`: The bootstrap server of your Redpanda cluster, including the port. For a self-hosted cluster, this is the address of the Kafka API listener, for example `redpanda.example.com:9092`. For Redpanda Cloud, use the bootstrap server URL from the **Overview** page. * `saslUsername` and `saslPassword`: The credentials of the SCRAM user. * `produceTopic`: The Redpanda topic that receives the shop floor data. * `consumeTopic`: The Redpanda 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: bootstrapServer: description: Bootstrap server of your Redpanda cluster type: string default: redpanda.example.com:9092 saslUsername: description: Username of the SCRAM user type: string default: connectware saslPassword: description: Password of the SCRAM user type: string produceTopic: description: Redpanda topic that receives the shop floor data type: string default: factory.machine-data consumeTopic: description: Redpanda 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 %} ### Producing Shop Floor Data to Redpanda To send data to Redpanda, we set up a `Cybus::Connection` resource with the `clientType` property set to `producer` and the `sasl` property set to the SCRAM mechanism of the Connectware user. In Connectware, the mechanism value is lowercase: `scram-sha-256` or `scram-sha-512`. It must match the mechanism that the user was created with. For all available properties, see [Kafka Connection Properties](/2-4-2/connectors/enterprise-connectors/kafka/kafkaconnection.md). {% code lineNumbers="true" %} ```yaml resources: redpandaProducerConnection: type: Cybus::Connection properties: protocol: Kafka connection: brokers: - !ref bootstrapServer clientType: producer sasl: mechanism: scram-sha-256 username: !ref saslUsername password: !ref saslPassword ``` {% endcode %} For a development cluster without authentication, omit the `sasl` property. Without `sasl` and without `mutualAuthentication`, the connection uses PLAINTEXT, unencrypted and unauthenticated. Do not use this variant in production. {% code lineNumbers="true" %} ```yaml connection: brokers: - !ref bootstrapServer clientType: producer ``` {% endcode %} A write endpoint addresses the Redpanda 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 redpandaProducerConnection 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 Redpanda 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 Redpanda to keep all records of one machine on the same partition. ### Consuming a Redpanda Topic into MQTT For the opposite direction, for example to receive commands or analytics results from other Redpanda 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 Redpanda 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 redpandaConsumerConnection: type: Cybus::Connection properties: protocol: Kafka connection: brokers: - !ref bootstrapServer clientType: consumer groupId: !ref consumerGroupId sasl: mechanism: scram-sha-256 username: !ref saslUsername password: !ref saslPassword commandsEndpoint: type: Cybus::Endpoint properties: protocol: Kafka connection: !ref redpandaConsumerConnection subscribe: topic: !ref consumeTopic fromBeginning: false commandsMapping: type: Cybus::Mapping properties: mappings: - subscribe: endpoint: !ref commandsEndpoint publish: topic: !sub '${topicRoot}/redpanda/commands' rules: - transform: expression: 'value.message' ``` {% endcode %} Every record that another client produces to the consume topic now appears on the `enterprise/redpanda/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 from your Redpanda 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 if SASL is configured but the Redpanda listener does not serve TLS, 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. Consume the produce topic and check for the new record, either in the topic view of the Redpanda Console or with the `rpk` command line client: 
rpk topic consume factory.machine-data --offset end
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. If Redpanda rejects a request, for example because the topic does not exist or an ACL is missing, the message on the `/res` topic contains an `error` property. 5. Produce a test record to the consume topic and check that it appears on the `enterprise/redpanda/commands` MQTT topic in the Data Explorer: 
echo '{"command":"set-speed","value":80}' | rpk topic produce factory.commands
## Service Commissioning File Example {% file src="/files/I1MpUsXeZ3TRt58F3UXf" %} {% code title="redpanda-example.yml" lineNumbers="true" expandable="true" %} ```yaml --- # ---------------------------------------------------------------------------- # Service Commissioning File # ---------------------------------------------------------------------------- # Copyright: Cybus GmbH # Contact: support@cybus.io # ---------------------------------------------------------------------------- # Redpanda Integration (Example) # ---------------------------------------------------------------------------- description: > Service commissioning file for the bidirectional integration between Connectware and Redpanda (Example) metadata: name: Redpanda Integration provider: cybus homepage: https://www.cybus.io version: 1.0.0 parameters: bootstrapServer: description: Bootstrap server of your Redpanda cluster type: string default: redpanda.example.com:9092 saslUsername: description: Username of the SCRAM user type: string default: connectware saslPassword: description: Password of the SCRAM user type: string produceTopic: description: Redpanda topic that receives the shop floor data type: string default: factory.machine-data consumeTopic: description: Redpanda 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 to Redpanda using SASL/SCRAM over TLS # The mechanism must match the mechanism the user was created with # (scram-sha-256 or scram-sha-512) # Remove the sasl property to connect to an unsecured development # cluster over PLAINTEXT redpandaProducerConnection: type: Cybus::Connection properties: protocol: Kafka connection: brokers: - !ref bootstrapServer clientType: producer sasl: mechanism: scram-sha-256 username: !ref saslUsername password: !ref saslPassword # Produces shop floor data to the Redpanda topic machineDataEndpoint: type: Cybus::Endpoint properties: protocol: Kafka connection: !ref redpandaProducerConnection 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($) } ] }' # Consumer connection to Redpanda # A Kafka connection is either a producer or a consumer, never both redpandaConsumerConnection: type: Cybus::Connection properties: protocol: Kafka connection: brokers: - !ref bootstrapServer clientType: consumer groupId: !ref consumerGroupId sasl: mechanism: scram-sha-256 username: !ref saslUsername password: !ref saslPassword # Consumes the Redpanda topic into the MQTT topic hierarchy commandsEndpoint: type: Cybus::Endpoint properties: protocol: Kafka connection: !ref redpandaConsumerConnection subscribe: topic: !ref consumeTopic fromBeginning: false commandsMapping: type: Cybus::Mapping properties: mappings: - subscribe: endpoint: !ref commandsEndpoint publish: topic: !sub '${topicRoot}/redpanda/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/redpanda-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.