Confluent Ingress

Confluent Logo

Nstream provides a Confluent Adapter library that simplifies connectivity to Confluent Cloud This guide demonstrates how to consume from Confluent topics and process responses in Web Agents using minimal boilerplate.




implementation 'io.nstream:nstream-adapter-confluent:'




Ideal ConfluentIngestingPatch Conditions


Then you simply need to use the ConfluentIngestingPatch class, often without overriding any methods. Let’s demonstrate this by recreating the ConfluentConsumerAgent example from the open source documentation wholly in server.recon:

The Confluent Adapter library is designed to simplify the connectivity to Confluent Cloud. To connect you need your Confluent Cloud bootstrap.servers, api.key and api-secret. The underlying adapter is very similar to the Kafka Adapter and will have identical behaviors because they share common code. The Confluent Adapter library will automatically add additional properties that are required for the Kafka client will need to connect to Confluent Cloud.

Protect your secrets

Make sure that you protect your secrets.
We suggest storing your credentials in a Kubernetes Secret

# required properties
ccloud.bootstrap.servers=<ccloud bootstrap.servers>
ccloud.api.key=<ccloud api key>
ccloud.api.secret=<ccloud api secret>

# the above settings will be used to generate the configuration needed 
# to start a Java Kafka Client.

# Any setting with kafka. will be overridden before the underlying Java 
# Kafka client is created.
# server.recon
provisions: {
  @provision("consumer-properties") {
    class: "nstream.adapter.common.provision.PropertiesProvision",
    # Confluent consumer properties go here
    config: "/secrets/"

vehicle: @fabric {
  @plane(class: "nstream.adapter.runtime.AppPlane")
  # ConfluentConsumingAgent replacement
  @node {
    uri: "/confluent"
    @agent(class: "nstream.adapter.confluent.ConfluentIngestingPatch") {
      # ConfluentIngressSettings values go here
      confluentIngressConf: @confluentIngressSettings {
        consumerPropertiesProvisionName: "consumer-properties"
        name: test-consumer
        topics: {"schema-topic"}
        valueContentTypeOverride: "json"
        relaySchema: @command {
          nodeUri: {
            $key # $value$id will also work here
          laneUri: "addMessage"
          value: $value
  # VehicleAgent config can be copied directly, but you'll of course
  # need to implement VehicleAgent in Java.

# Configure desired web settings (e.g. port, websocket compression)