Confluent Cloud Integration

How to integrate Confluent Cloud with Connectware, including SASL authentication with API keys, producing shop floor data to a Confluent Cloud topic, and consuming a topic into MQTT.

This guide describes how to integrate Confluent Cloud with Connectware. You configure a service commissioning file that streams shop floor data to a Confluent Cloud topic through the Kafka connector and consumes a Confluent Cloud topic into the MQTT topic hierarchy in return. A complete example file is available at the end of this guide.

Objectives

• Connecting Connectware to a Confluent Cloud cluster with SASL authentication and a cluster API key.

• Producing shop floor data from the MQTT topic hierarchy to a Confluent Cloud topic.

• Consuming a Confluent Cloud topic into an MQTT topic.

Prerequisites

To follow this guide, you will need the following:

• A running instance of Cybus Connectware.

• A Confluent Cloud cluster, its bootstrap server address, and a cluster API key and secret. You find the bootstrap server under Cluster Settings in the Confluent Cloud Console and create API keys under API Keys.

• The Kafka topics that you want to produce to and consume from, created in your Confluent Cloud cluster. Confluent Cloud disables automatic topic creation by default, so producing to a topic that does not exist fails. For more information, see Manage Kafka Cluster Configuration Settings in Confluent Cloud.

• 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 Confluent Cloud Integration

Confluent Cloud is a managed Apache Kafka service. Clients connect to a bootstrap server of the form pkc-<id>.<region>.<provider>.confluent.cloud:9092 and must use TLS encryption and SASL authentication. With SASL PLAIN, the cluster API key is the username and the API secret is the password. For more information, see Connect Kafka Client Applications to Confluent Cloud.

Connectware communicates with Confluent Cloud 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. Avro or Protobuf serialization with the Confluent 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.

Confluent Cloud Connection Properties

The connection to Confluent Cloud requires the bootstrap server address and a cluster API key. 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 Confluent Cloud cluster, including the port. For example, pkc-00000.eu-central-1.aws.confluent.cloud:9092.

• confluentApiKey and confluentApiSecret: The cluster API key and secret.

• produceTopic: The Confluent Cloud topic that receives the shop floor data.

• consumeTopic: The Confluent Cloud 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 Confluent Cloud cluster
    type: string
    default: pkc-00000.eu-central-1.aws.confluent.cloud:9092

  confluentApiKey:
    description: Confluent Cloud cluster API key
    type: string

  confluentApiSecret:
    description: Confluent Cloud cluster API secret
    type: string

  produceTopic:
    description: Confluent Cloud topic that receives the shop floor data
    type: string
    default: factory.machine-data

  consumeTopic:
    description: Confluent Cloud 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

If the API key belongs to a service account with granular access control lists (ACLs), the service account needs write access to the producer topic, read access to the consumer topic, and read access to the consumer group. For more information, see ACLs Overview for Confluent Cloud.

Producing Shop Floor Data to Confluent Cloud

To send data to Confluent Cloud, we set up a Cybus::Connection resource with the clientType property set to producer and the sasl property set to the plain mechanism with the API key and secret. Confluent Cloud broker certificates are signed by a public certificate authority, so you do not need to configure the caCert or trustAllCertificates properties. For all available properties, see Kafka Connection Properties.

resources:
  confluentProducerConnection:
    type: Cybus::Connection
    properties:
      protocol: Kafka
      connection:
        brokers:
          - !ref bootstrapServer
        clientType: producer
        sasl:
          mechanism: plain
          username: !ref confluentApiKey
          password: !ref confluentApiSecret

A write endpoint addresses the Confluent Cloud 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 confluentProducerConnection
    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 Confluent Cloud topic as one Kafka 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 Confluent Cloud to keep all records of one machine on the same partition.

Consuming a Confluent Cloud Topic into MQTT

For the opposite direction, for example to receive commands or analytics results from other Confluent Cloud 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 Confluent Cloud 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.

confluentConsumerConnection:
  type: Cybus::Connection
  properties:
    protocol: Kafka
    connection:
      brokers:
        - !ref bootstrapServer
      clientType: consumer
      groupId: !ref consumerGroupId
      sasl:
        mechanism: plain
        username: !ref confluentApiKey
        password: !ref confluentApiSecret

commandsEndpoint:
  type: Cybus::Endpoint
  properties:
    protocol: Kafka
    connection: !ref confluentConsumerConnection
    subscribe:
      topic: !ref consumeTopic
      fromBeginning: false

commandsMapping:
  type: Cybus::Mapping
  properties:
    mappings:
      - subscribe:
          endpoint: !ref commandsEndpoint
        publish:
          topic: !sub '${topicRoot}/confluent/commands'
        rules:
          - transform:
              expression: 'value.message'

Every record that another client produces to the consume topic now appears on the enterprise/confluent/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 Confluent Cloud cluster.

2. Check that both connections are in the Connected state on the service details page in the Admin UI. If the API key or secret is wrong, 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. Open the topic in the Confluent Cloud Console and check the Messages tab for the new record. The result of every produce request is also published to the /res topic of the endpoint. Use the Data Explorer to inspect it. If Confluent Cloud rejects a request, for example because the topic does not exist, the message on the /res topic contains an error property.

5. Produce a test record to the consume topic in the Confluent Cloud Console and check that it appears on the enterprise/confluent/commands MQTT topic in the Data Explorer.

Service Commissioning File Example

confluent-cloud-example.yml

4KB

Open

confluent-cloud-example.yml

---
# ----------------------------------------------------------------------------
# Service Commissioning File
# ----------------------------------------------------------------------------
# Copyright: Cybus GmbH
# Contact: [e-mail]
# ----------------------------------------------------------------------------
# Confluent Cloud Integration (Example)
# ----------------------------------------------------------------------------

description: >
  Service commissioning file for the bidirectional integration between
  Connectware and Confluent Cloud (Example)

metadata:
  name: Confluent Cloud Integration
  provider: cybus
  homepage: https://www.cybus.io
  version: 1.0.0

parameters:
  bootstrapServer:
    description: Bootstrap server of your Confluent Cloud cluster
    type: string
    default: pkc-00000.eu-central-1.aws.confluent.cloud:9092

  confluentApiKey:
    description: Confluent Cloud cluster API key
    type: string

  confluentApiSecret:
    description: Confluent Cloud cluster API secret
    type: string

  produceTopic:
    description: Confluent Cloud topic that receives the shop floor data
    type: string
    default: factory.machine-data

  consumeTopic:
    description: Confluent Cloud 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 Confluent Cloud using SASL PLAIN over TLS
  confluentProducerConnection:
    type: Cybus::Connection
    properties:
      protocol: Kafka
      connection:
        brokers:
          - !ref bootstrapServer
        clientType: producer
        sasl:
          mechanism: plain
          username: !ref confluentApiKey
          password: !ref confluentApiSecret

  # Produces shop floor data to the Confluent Cloud topic
  machineDataEndpoint:
    type: Cybus::Endpoint
    properties:
      protocol: Kafka
      connection: !ref confluentProducerConnection
      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 Confluent Cloud
  # A Kafka connection is either a producer or a consumer, never both
  confluentConsumerConnection:
    type: Cybus::Connection
    properties:
      protocol: Kafka
      connection:
        brokers:
          - !ref bootstrapServer
        clientType: consumer
        groupId: !ref consumerGroupId
        sasl:
          mechanism: plain
          username: !ref confluentApiKey
          password: !ref confluentApiSecret

  # Consumes the Confluent Cloud topic into the MQTT topic hierarchy
  commandsEndpoint:
    type: Cybus::Endpoint
    properties:
      protocol: Kafka
      connection: !ref confluentConsumerConnection
      subscribe:
        topic: !ref consumeTopic
        fromBeginning: false

  commandsMapping:
    type: Cybus::Mapping
    properties:
      mappings:
        - subscribe:
            endpoint: !ref commandsEndpoint
          publish:
            topic: !sub '${topicRoot}/confluent/commands'
          rules:
            - transform:
                expression: 'value.message'