> 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/confluent-cloud-integration.md). # Confluent Cloud Integration 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](https://docs.confluent.io/cloud/current/clusters/broker-config.html). * 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 Confluent Cloud Integration Confluent Cloud is a managed Apache Kafka service. Clients connect to a bootstrap server of the form `pkc-...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](https://docs.confluent.io/cloud/current/client-apps/config-client.html). Connectware communicates with Confluent Cloud 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. 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 (`////`). 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`. {% code lineNumbers="true" %} ```yaml 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 ``` {% endcode %} {% hint style="info" %} 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](https://docs.confluent.io/cloud/current/security/access-control/acls/overview.html). {% endhint %} ### 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](/2-4-2/connectors/enterprise-connectors/kafka/kafkaconnection.md). {% code lineNumbers="true" %} ```yaml resources: confluentProducerConnection: type: Cybus::Connection properties: protocol: Kafka connection: brokers: - !ref bootstrapServer clientType: producer sasl: mechanism: plain username: !ref confluentApiKey password: !ref confluentApiSecret ``` {% endcode %} 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](/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 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($) } ] }' ``` {% endcode %} 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: {% 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 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](/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 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' ``` {% endcode %} 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](/2-4-2/monitoring/data-explorer.md) 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 {% file src="/files/VrVbZCW4PKvSaErndQoY" %} {% code title="confluent-cloud-example.yml" lineNumbers="true" expandable="true" %} ```yaml --- # ---------------------------------------------------------------------------- # Service Commissioning File # ---------------------------------------------------------------------------- # Copyright: Cybus GmbH # Contact: support@cybus.io # ---------------------------------------------------------------------------- # 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' ``` {% 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/confluent-cloud-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.