Confluent Cloud with Schema Registry Configuration Example

This production‑ready configuration example targets Confluent Cloud, a fully managed Apache Kafka service. It uses the KafkaSender with specific security settings for Confluent Cloud and includes Schema Registry integration.

This Confluent schema registry configuration example uses a local FileStorageProvider for large blob and slob data, demonstrating a hybrid approach with cloud messaging and local storage. It includes file and console logging for monitoring in a production environment.

{
   "Modules" : {
      "Reader" : "File",
      "Sender" : "Kafka",
      "TopicManager" : "File",
      "Serializer" : "Json",
      "StorageProvider" : "File",
      "SchemaGenerator" : "Json", 
      "SchemaRegistry" : "Confluent"
   },
   "Configurations" : {
      "KafkaSender" : {
         "Properties" : {
            "bootstrap.servers" : "your-kafka-server.confluent.cloud:9092",
            "security.protocol" : "SASL_SSL",
            "sasl.mechanisms" : "PLAIN",
            "sasl.username" : "<API_KEY>",
            "sasl.password" : "<API_SECRET>",
            "schema.registry.url" :
                 "https://your-schema-registry.confluent.cloud",
            "schema.registry.basic.auth.user.info" : "<SR_API_KEY>:
                 <SR_API_SECRET>",
            "basic.auth.credentials.source" : "USER_INFO"
         }
      },
      "Loggers" : [
         {
            "LogFileName" : "C:\\Events\\Logs\\JadeEventProducer_Info.log",
            "MaxLevel" : "Info",
            "Delay" : 1000,
            "RotateSize" : 100,
            "RotateKeep" : 10,
            "Sink" : "File"
         },
         {
            "MaxLevel" : "Warn",
            "Sink" : "Console"
         }
      ],
      "FileReader" : {
         "CaptureDirectory" : "C:\\Events\\ChangeCapture",
         "UndeliverableDirectory" : "C:\\Events\\ChangeCapture\\undeliverable",
         "Delay" : 1000,
         "RunContinuously" : true
      },
      "FileTopicManager" : {
         "TopicFileName" : "C:\\Events\\Config\\JadeTopics.config.json"
      },
      "FileStorageProvider" : {
         "RootDirectory" : "C:\\Events\\ChangeCapture\\Blobs"
      },
      "SizeThresholds" : {
         "SlobThresholdBytes" : "1MB",
         "BlobThresholdBytes" : "5MB"
      },
      "ConfluentSchemaRegistry" : {
         "Endpoint" : "https://endpoint.confluent.cloud",
         "SchemaFormat" : "Json",
         "Authentication" : {
           "Type" : "ApiKey",
           "ApiKeyName" : "1234567890",
           "ApiKeyValue" : "ABCDEFEGGHIJKLMNOPQRSTUVWXYZ1234567890+"
         }
      },
      "Catalog" : {
         "CatalogDirectory" : "C:\\Jade\\System\\Journals"
      },
      "SchemaCache" : {
         "CacheRootDirectory" : "C:\\Events\\SchemaCache"
      }
   }
}

Key aspects of this Confluent Cloud with SchemaRegistry configuration FileSender example are listed in the following table.

Element	Description
KafkaSender	Configured for Confluent Cloud with security settings
ConfluentSchemaRegistry	Includes configuration for Confluent Cloud
Loggers	Contains a configuration for two Loggers: one for detailed file logging and another for console warnings
FileStorageProvider	Used for storing large blobs and slobs locally
SizeThresholds	Set to determine when to use external storage

OAuth2 on Confluent Cloud with Schema Registry

The following example shows the use of the ConfluentOAuth2 authentication type with a SchemaRegistry.

"ConfluentSchemaRegistry": {
         "Authentication":{
           "Type": "ConfluentOAuth2",
           "TenantId": "my-tenant-id",
           "ClientId": "my-client-id",
           "ClientSecret": "my-client-secret",
           "Scope": "<https://cluster‑id>.<region>.confluent.cloud/>",
           "IdentityPool": "pool-xxxxx",
           "TargetCluster" : "lsrc-yyyyyy"
}

The IdentityPool and TargetCluster parameters are required to use the Confluent SchemaRegistry with OAuth2. These parameters configure the Confluent Cloud SchemaRegistry to use the associated Azure Active Directory (AD) identity provider. For details, see "Authentication Types", later in this chapter.