Redpanda Integration

How to integrate Redpanda with Connectware, including producing shop floor data to a topic, consuming a topic into MQTT, and securing the connection with SASL/SCRAM.

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

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

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). Connectware communicates with Redpanda through the Kafka connector, with no Redpanda-specific configuration. If you use Confluent Cloud instead, see Confluent Cloud Integration.

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

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 command:

rpk topic create factory.machine-data factory.commands

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. For production, enable SASL/SCRAM authentication. Create a superuser, add it to the superusers cluster property, and enable SASL on the Kafka API:

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

• ${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.

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

• ${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.

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

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

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

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.

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

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.

connection:
  brokers:
    - !ref bootstrapServer
  clientType: producer

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

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($) } ] }'

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:

{
  "temperature": 42.1,
  "pressure": 5.4,
  "serial": "SN-0001"
}

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

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'

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

redpanda-example.yml

4KB

Open

redpanda-example.yml

---
# ----------------------------------------------------------------------------
# Service Commissioning File
# ----------------------------------------------------------------------------
# Copyright: Cybus GmbH
# Contact: [e-mail]
# ----------------------------------------------------------------------------
# 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'