Evolving an Avro Schema in a Kafka Application

1. Clone the starter project into your home directory:

cd ~/
git clone https://github.com/linuxacademy/content-ccdak-schema-evolve-lab.git

1. Run the code to ensure it works before modifying it:

cd content-ccdak-schema-evolve-lab
./gradlew runProducer
./gradlew runConsumer

Note: The consumer should output some records that were created by the producer.

1. Edit the schema definition file:

vi src/main/avro/com/linuxacademy/ccdak/schemaregistry/Purchase.avsc

1. Add the member_id field with a blank default:

{
  "namespace": "com.linuxacademy.ccdak.schemaregistry",
  "compatibility": "FORWARD",
  "type": "record",
  "name": "Purchase",
  "fields": [
    {"name": "id", "type": "int"},
    {"name": "product", "type": "string"},
    {"name": "quantity", "type": "int"},
    {"name": "member_id", "type": "int", "default": 0}
  ]
}

1. Edit the schema definition file:

vi src/main/avro/com/linuxacademy/ccdak/schemaregistry/Purchase.avsc

1. Add the member_id field with a blank default:

{
  "namespace": "com.linuxacademy.ccdak.schemaregistry",
  "compatibility": "FORWARD",
  "type": "record",
  "name": "Purchase",
  "fields": [
    {"name": "id", "type": "int"},
    {"name": "product", "type": "string"},
    {"name": "quantity", "type": "int"},
    {"name": "member_id", "type": "int", "default": 0}
  ]
}

Update the Producer to Set the member_id for the Records It Publishes

1. Edit the producer Main class:

vi src/main/java/com/linuxacademy/ccdak/schemaregistry/ProducerMain.java

1. Implement the new member_id field in the producer by setting it for the records being produced:

package com.linuxacademy.ccdak.schemaregistry;

import io.confluent.kafka.serializers.AbstractKafkaAvroSerDeConfig;
import io.confluent.kafka.serializers.KafkaAvroSerializer;
import java.util.Properties;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

public class ProducerMain {

    public static void main(String[] args) {
        final Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.RETRIES_CONFIG, 0);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
        props.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081");

        KafkaProducer<String, Purchase> producer = new KafkaProducer<String, Purchase>(props);

        Purchase apples = new Purchase(1, "apples", 17, 77543);
        producer.send(new ProducerRecord<String, Purchase>("inventory_purchases", apples.getId().toString(), apples));

        Purchase oranges = new Purchase(2, "oranges", 5, 56878);
        producer.send(new ProducerRecord<String, Purchase>("inventory_purchases", oranges.getId().toString(), oranges));

        producer.close();
    }

}

1. Run the producer:

./gradlew runProducer

1. Run the consumer:

./gradlew runConsumer

1. Verify the data in the output file. We should see the new member_id data in the last lines of the file:

cat /home/cloud_user/output/output.txt