Skip to main content

Use schema registry with Java producers and consumers

Aiven for Apache Kafka® provides schema registry functionality through Karapace. Karapace lets you store, retrieve, and evolve schemas without rebuilding producer or consumer code.

The examples use Avro. For Protobuf or JSON Schema, generate the classes first, then apply the same connection and authentication settings.

Workflow overview​

To produce and consume Avro messages in Java using the schema registry:

  1. Define your Avro schema.
  2. Generate Java classes from the schema.
  3. Add the required Maven dependencies.
  4. Optional: Create a keystore, and create a truststore only if you use SASL authentication.
  5. Configure your Kafka producer and consumer properties.

Prerequisites​

Get connection details​

On the service Overview page, open Connection information.

  1. On the Apache Kafka tab, copy the Service URI for the bootstrap servers.
  2. On the Schema Registry tab, copy the Service URI, User, and Password.

Variables​

Replace the following placeholders in the example configuration:

VariableDescription
BOOTSTRAPSERVERSKafka service URI from Connection information on the service overview page
KEYSTOREPath to the keystore file
KEYSTOREPASSWORDPassword for the keystore
TRUSTSTOREPath to the truststore file
TRUSTSTOREPASSWORDPassword for the truststore
SSLKEYPASSWORDPassword for the private key in the keystore
SCHEMAREGISTRYURLSchema registry URI from Connection information
SCHEMAREGISTRYUSERSchema registry username from Connection information
SCHEMAREGISTRYPASSWORDSchema registry password from Connection information
TOPIC_NAMEKafka topic name

Define an Avro schema​

Create an Avro schema file. For example, save the following schema in a file named ClickRecord.avsc:

{
  "type": "record",
  "name": "ClickRecord",
  "namespace": "io.aiven.avro.example",
  "fields": [
    {"name": "session_id", "type": "string"},
    {"name": "browser", "type": ["string", "null"]},
    {"name": "campaign", "type": ["string", "null"]},
    {"name": "channel", "type": "string"},
    {"name": "referrer", "type": ["string", "null"], "default": "None"},
    {"name": "ip", "type": ["string", "null"]}
  ]
}

This schema defines a record named ClickRecord in the namespace io.aiven.avro.example. The record has the fields session_id, browser, campaign, channel, referrer, and ip.

Generate Java classes and add dependencies​

Generate Java classes from your schema, then add the required dependencies to your pom.xml:

Configure producer and consumer properties​

For complete example code, see the Aiven examples GitHub repository.

Producer configuration​

props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAPSERVERS);
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");
props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, TRUSTSTORE);
props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, TRUSTSTOREPASSWORD);
props.put(SslConfigs.SSL_KEYSTORE_TYPE_CONFIG, "PKCS12");
props.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, KEYSTORE);
props.put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, KEYSTOREPASSWORD);
props.put(SslConfigs.SSL_KEY_PASSWORD_CONFIG, SSLKEYPASSWORD);
props.put("schema.registry.url", SCHEMAREGISTRYURL);
props.put("basic.auth.credentials.source", "USER_INFO");
props.put("basic.auth.user.info", SCHEMAREGISTRYUSER + ":" + SCHEMAREGISTRYPASSWORD);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class.getName());

Consumer configuration​

props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAPSERVERS);
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");
props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, TRUSTSTORE);
props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, TRUSTSTOREPASSWORD);
props.put(SslConfigs.SSL_KEYSTORE_TYPE_CONFIG, "PKCS12");
props.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, KEYSTORE);
props.put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, KEYSTOREPASSWORD);
props.put(SslConfigs.SSL_KEY_PASSWORD_CONFIG, SSLKEYPASSWORD);
props.put("schema.registry.url", SCHEMAREGISTRYURL);
props.put("basic.auth.credentials.source", "USER_INFO");
props.put("basic.auth.user.info", SCHEMAREGISTRYUSER + ":" + SCHEMAREGISTRYPASSWORD);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class.getName());
props.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, true);
props.put(ConsumerConfig.GROUP_ID_CONFIG, "clickrecord-example-group");

Replace the placeholders with the values from the variables section.

Related pages