60-Day Kafka 4 Learning Plan · Week 4 Lab — Schema & Serialization Sources: Kafka: The Definitive Guide Ch.3 & 9 · avro.apache.org · docs.confluent.io/schema-registry
Goal
Execute a zero-downtime migration of the labs.events topic from JSON to Avro using a dual-write strategy, consumer group cutover, a tested rollback path, and verification that actually verifies — including fixing a gap in the naive test setup.
This is Phase 2 of the Kafka 4 learning series. Phase 1 built a producer-consumer-WebSocket pipeline with a JSON LabEvent flowing through the lab-events topic.
Phase 2 migrates that topic from JSON to Avro without ever stopping the consumers.
This is a lab. It is written to simulate a real migration decision with real operational constraints: no topic rename, no downtime, no forced consumer restart. Everything happens incrementally, with a rollback path at every phase.
The Problem with JSON Topics at Scale
JSON works well until the day it doesn’t.
A team adds a field. Another team doesn’t deploy their consumer for two weeks. A third service has been ignoring that field silently. Nobody notices until production drops an event that doesn’t match the deserialization code somewhere downstream.
JSON gives you flexibility. It gives you zero schema enforcement. Every consumer re-implements the same deserialization logic. Every breaking change is invisible until runtime. There is no contract.
Avro fixes this. The schema is the contract. The Schema Registry enforces it. Incompatible changes are rejected before they reach a topic. Consumers know the exact shape of the data before they read the first byte.
The migration from JSON to Avro is the hard part. This post documents how to do it safely.
The Four-Phase Plan
Phase 1 — Prepare Create lab-events.v2 topic. Register Avro schema. Add dependencies.
Phase 2 — Dual-write Producer writes to both topics simultaneously. Avro flag is off initially.
Phase 3 — Cutover Start Avro consumer. Enable flag. Monitor both consumer groups at lag=0.
Phase 4 — Clean up Remove JSON consumer. Remove JSON producer path. Delete lab-events.Each phase is independently deployable. Rollback from any phase requires only toggling the feature flag or redeploying the previous artifact — no topic deletion, no data loss.
Phase 1: Prepare
1.1 Docker Compose — Add Schema Registry
Schema Registry needs its own Kafka listener. The existing docker-compose uses apache/kafka:4.0.0 with a single PLAINTEXT listener advertised to localhost:9092. That works for applications running outside Docker, but Schema Registry runs inside the Docker network and cannot reach localhost.
The fix is a second listener — INTERNAL on port 29092 — advertised as kafka-1:29092. All containers (labs-api, labs-socket, schema-registry) use kafka-1:29092. External clients (IDE, curl, CLI tools) still use localhost:9092.
kafka-1:
image: apache/kafka:4.0.0
environment:
KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,INTERNAL://0.0.0.0:29092,CONTROLLER://0.0.0.0:9093
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092,INTERNAL://kafka-1:29092
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,INTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT
KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
# ...Schema Registry connects on the internal listener:
schema-registry:
image: confluentinc/cp-schema-registry:7.6.0
depends_on:
kafka-1:
condition: service_healthy
ports:
- "8081:8081"
environment:
SCHEMA_REGISTRY_HOST_NAME: schema-registry
SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: kafka-1:29092
SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081
SCHEMA_REGISTRY_KAFKASTORE_TOPIC_REPLICATION_FACTOR: 1Port conflict: labs-socket previously mapped to 8081:8081 on the host. Schema Registry now occupies host port 8081. Move labs-socket to 9081:8081 — internal port unchanged, only the host binding changes.
1.2 Avro Schema
Create labs-api/src/main/avro/lab-event.avsc:
{
"type": "record",
"name": "LabEvent",
"namespace": "com.labs.events",
"doc": "Avro schema for LabEvent — Phase 2 migration target of lab-events JSON topic.",
"fields": [
{ "name": "eventId", "type": "string" },
{ "name": "type", "type": "string" },
{ "name": "payload", "type": "string" },
{ "name": "timestamp", "type": "long",
"doc": "Epoch milliseconds. Convert with Instant.ofEpochMilli()." },
{ "name": "source", "type": "string" },
{ "name": "couponCode", "type": ["null", "string"], "default": null }
]
}Three design decisions worth noting:
Namespace: com.labs.events is different from the application package com.boottechsolutions. The generated class com.labs.events.LabEvent is distinct from the domain record com.boottechsolutions.labsapi.model.LabEvent. No naming collision, no refactoring required in either service.
Timestamp as long: Instant maps to {"type": "long", "logicalType": "timestamp-millis"} in full Avro. For this lab, using plain long avoids needing to configure the Avro Maven plugin for java.time support (which varies across plugin versions). Store as timestamp.toEpochMilli(), read back as Instant.ofEpochMilli(event.getTimestamp()).
couponCode as union ["null", "string"]: Avro requires explicit null union for nullable fields. The default: null must appear when null is the first type in the union. Swap the order (["string", "null"]) and the default changes to a required non-null value — a common mistake.
Copy the same .avsc file to labs-socket/src/main/avro/ — both services generate the same com.labs.events.LabEvent class independently from the same schema. In a mature setup, the schema lives in a shared library or is fetched from the registry at build time. For this lab, the copy approach is sufficient.
1.3 Maven Dependencies
Add to both labs-api/pom.xml and labs-socket/pom.xml:
<properties>
<avro.version>1.11.3</avro.version>
<confluent.version>7.6.0</confluent.version>
</properties>
<repositories>
<repository>
<id>confluent</id>
<url>https://packages.confluent.io/maven/</url>
</repository>
</repositories>
<dependencies>
<dependency>
<groupId>org.apache.avro</groupId>
<artifactId>avro</artifactId>
<version>${avro.version}</version>
</dependency>
<dependency>
<groupId>io.confluent</groupId>
<artifactId>kafka-avro-serializer</artifactId>
<version>${confluent.version}</version>
<exclusions>
<exclusion>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
</exclusion>
</exclusions>
</dependency>
<!-- test scope only -->
<dependency>
<groupId>io.confluent</groupId>
<artifactId>kafka-schema-registry-client</artifactId>
<version>${confluent.version}</version>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.avro</groupId>
<artifactId>avro-maven-plugin</artifactId>
<version>${avro.version}</version>
<executions>
<execution>
<phase>generate-sources</phase>
<goals><goal>schema</goal></goals>
<configuration>
<sourceDirectory>${project.basedir}/src/main/avro</sourceDirectory>
<outputDirectory>
${project.build.directory}/generated-sources/avro
</outputDirectory>
<stringType>String</stringType>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>The Jackson exclusion prevents the Confluent artifact from pulling an older jackson-databind version that conflicts with Spring Boot’s managed version.
After adding the plugin, run mvn generate-sources to confirm the class is generated at target/generated-sources/avro/com/labs/events/LabEvent.java before writing any producer or consumer code.
1.4 Create lab-events.v2 Topic and Register Schema
Start the updated docker-compose stack, then run the migration script:
docker compose up -d
./scripts/migrate.shThe script:
- Reads the partition count of
lab-eventsand createslab-events.v2with the same count - Reads
lab-event.avsc, wraps it in the Schema Registry JSON envelope, and POSTs to/subjects/lab-event-value/versions - Prints the assigned schema ID and verifies retrieval
Why match partition counts? Kafka routes messages to partitions via murmur2(key) % numPartitions. If both topics have the same partition count, a given eventId key always lands on the same partition number in both topics. Consumers that process related events sequentially (order matters) retain that guarantee across the migration window.
Why register the schema before enabling dual-write? If the producer registers the schema at first send, that send incurs a synchronous HTTP round-trip to Schema Registry. Pre-registering eliminates that latency spike from production traffic.
Verify the schema is registered:
curl -s http://localhost:8081/subjects/lab-event-value/versions/latest | python3 -m json.tool
Phase 2: Dual-Write Producer
2.1 Two KafkaTemplate Beans
The existing KafkaProducerConfig has one @Bean KafkaTemplate<String, LabEvent> wired with JsonSerializer. Add a second bean for Avro:
@Configuration
public class KafkaProducerConfig {
@Value("${SCHEMA_REGISTRY_URL:http://localhost:8081}")
private String schemaRegistryUrl;
// ─── Existing JSON producer ───────────────────────────────────────────
@Bean
@Primary
public ProducerFactory<String, LabEvent> jsonProducerFactory(KafkaProperties props) {
Map<String, Object> config = new HashMap<>(props.buildProducerProperties(null));
config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
config.put(ProducerConfig.ACKS_CONFIG, "all");
config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
config.put(JsonSerializer.ADD_TYPE_INFO_HEADERS, false);
return new DefaultKafkaProducerFactory<>(config);
}
@Bean("jsonKafkaTemplate")
@Primary
public KafkaTemplate<String, LabEvent> jsonKafkaTemplate(
ProducerFactory<String, LabEvent> jsonProducerFactory) {
return new KafkaTemplate<>(jsonProducerFactory);
}
// ─── New Avro producer ────────────────────────────────────────────────
@Bean
public ProducerFactory<String, com.labs.events.LabEvent> avroProducerFactory(KafkaProperties props) {
Map<String, Object> config = new HashMap<>(props.buildProducerProperties(null));
config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
config.put(ProducerConfig.ACKS_CONFIG, "all");
config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
config.put(KafkaAvroSerializerConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);
config.put(KafkaAvroSerializerConfig.AUTO_REGISTER_SCHEMAS, true);
return new DefaultKafkaProducerFactory<>(config);
}
@Bean("avroKafkaTemplate")
public KafkaTemplate<String, com.labs.events.LabEvent> avroKafkaTemplate(
ProducerFactory<String, com.labs.events.LabEvent> avroProducerFactory) {
return new KafkaTemplate<>(avroProducerFactory);
}
}@Primary on the JSON beans tells Spring’s autowiring which KafkaTemplate to inject when there’s no @Qualifier. The Avro beans are always injected by name.
2.2 Feature-Flag Dual-Write
@Service
@Slf4j
public class EventProducer {
private final KafkaTemplate<String, LabEvent> jsonKafka;
private final KafkaTemplate<String, com.labs.events.LabEvent> avroKafka;
@Value("${kafka.topic.lab-events:lab-events}")
private String jsonTopic;
@Value("${kafka.topic.lab-events-v2:lab-events.v2}")
private String avroTopic;
@Value("${feature.avro-enabled:false}")
private boolean avroEnabled;
public EventProducer(
@Qualifier("jsonKafkaTemplate") KafkaTemplate<String, LabEvent> jsonKafka,
@Qualifier("avroKafkaTemplate") KafkaTemplate<String, com.labs.events.LabEvent> avroKafka) {
this.jsonKafka = jsonKafka;
this.avroKafka = avroKafka;
}
public void publish(LabEvent event) {
sendJson(event);
if (avroEnabled) {
sendAvro(event);
}
}
private void sendAvro(LabEvent event) {
com.labs.events.LabEvent avroEvent = toAvro(event);
avroKafka.send(avroTopic, avroEvent.getEventId(), avroEvent)
.whenComplete((result, ex) -> {
if (ex != null) {
log.error("Avro send failed: eventId={} — JSON topic still has the event",
event.eventId(), ex);
}
});
}
private com.labs.events.LabEvent toAvro(LabEvent src) {
return com.labs.events.LabEvent.newBuilder()
.setEventId(src.eventId())
.setType(src.type())
.setPayload(src.payload())
.setTimestamp(src.timestamp().toEpochMilli())
.setSource(src.source())
.setCouponCode(src.couponCode())
.build();
}
}Add to application.yml:
feature:
avro-enabled: ${FEATURE_AVRO_ENABLED:false}Deploy labs-api at this point. The feature flag is false. Nothing changes in production yet — the Avro code path exists but is inactive.
2.3 The Dual-Write Correctness Gap
The two send() calls are not atomic. The following scenario is possible:
1. jsonKafka.send("lab-events", event) → SUCCESS, offset 9042
2. avroKafka.send("lab-events.v2", event) → FAILS (Schema Registry timeout)lab-events.v2 now has one fewer message than lab-events. The consumer on lab-events.v2 will never see event 9042.
This is the dual-write correctness gap. How much it matters depends on the event type.
For this migration window: accept the gap. The sendAvro() failure is logged as an error (not thrown), so the HTTP response to the caller succeeds. Run verify-parity.sh to detect any drift. A gap of 1-2 events during Schema Registry restart is recoverable — a sustained gap of hundreds means something is structurally wrong.
For financial or audit events: the transactional outbox pattern eliminates the gap. The producer writes to a database outbox table in the same transaction as the business operation. A relay process reads the outbox and sends to both Kafka topics in a retry loop. The source of truth is the database, not the in-flight Kafka send.
For lab-events, the correctness gap is acceptable.
Phase 3: Cutover
3.1 Avro Consumer Factory in labs-socket
Add a second consumer factory alongside the existing JSON factory:
@Configuration
@EnableKafka
public class KafkaConsumerConfig {
@Value("${SCHEMA_REGISTRY_URL:http://localhost:8081}")
private String schemaRegistryUrl;
// ─── Existing JSON consumer (@Primary — used by default) ─────────────
@Bean
@Primary
public ConsumerFactory<String, LabEvent> jsonConsumerFactory(KafkaProperties props) {
Map<String, Object> config = new HashMap<>(props.buildConsumerProperties(null));
config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
config.put(JsonDeserializer.TRUSTED_PACKAGES, "com.boottechsolutions.labssocket.model");
config.put(JsonDeserializer.USE_TYPE_INFO_HEADERS, false);
config.put(JsonDeserializer.VALUE_DEFAULT_TYPE, LabEvent.class.getName());
return new DefaultKafkaConsumerFactory<>(config, new StringDeserializer(),
new JsonDeserializer<>(LabEvent.class, false));
}
@Bean("kafkaListenerContainerFactory")
@Primary
public ConcurrentKafkaListenerContainerFactory<String, LabEvent> kafkaListenerContainerFactory(
ConsumerFactory<String, LabEvent> jsonConsumerFactory) {
var factory = new ConcurrentKafkaListenerContainerFactory<String, LabEvent>();
factory.setConsumerFactory(jsonConsumerFactory);
factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
return factory;
}
// ─── New Avro consumer ────────────────────────────────────────────────
@Bean("avroConsumerFactory")
public ConsumerFactory<String, com.labs.events.LabEvent> avroConsumerFactory(KafkaProperties props) {
Map<String, Object> config = new HashMap<>(props.buildConsumerProperties(null));
config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class);
config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
config.put(KafkaAvroDeserializerConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);
// true = use generated LabEvent SpecificRecord, not GenericRecord
config.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, true);
return new DefaultKafkaConsumerFactory<>(config, new StringDeserializer(),
new KafkaAvroDeserializer<>());
}
@Bean("avroKafkaListenerContainerFactory")
public ConcurrentKafkaListenerContainerFactory<String, com.labs.events.LabEvent>
avroKafkaListenerContainerFactory(
@Qualifier("avroConsumerFactory")
ConsumerFactory<String, com.labs.events.LabEvent> avroConsumerFactory) {
var factory = new ConcurrentKafkaListenerContainerFactory<String, com.labs.events.LabEvent>();
factory.setConsumerFactory(avroConsumerFactory);
factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
return factory;
}
}3.2 AvroEventConsumer
@Component
@Slf4j
public class AvroEventConsumer {
private final SimpMessagingTemplate messaging;
public AvroEventConsumer(SimpMessagingTemplate messaging) {
this.messaging = messaging;
}
@KafkaListener(
topics = "${kafka.topic.lab-events-v2:lab-events.v2}",
groupId = "labs-socket-avro",
containerFactory = "avroKafkaListenerContainerFactory"
)
public void consume(com.labs.events.LabEvent event, Acknowledgment ack) {
log.info("Avro [labs-socket-avro] eventId={} type={} ts={}",
event.getEventId(), event.getType(),
Instant.ofEpochMilli(event.getTimestamp()));
try {
messaging.convertAndSend("/topic/events", toMessage(event));
ack.acknowledge();
} catch (Exception ex) {
log.error("Processing failed: eventId={}", event.getEventId(), ex);
// No ack — Spring Kafka redelivers
}
}
private AvroEventMessage toMessage(com.labs.events.LabEvent event) {
return new AvroEventMessage(
event.getEventId(), event.getType(), event.getPayload(),
Instant.ofEpochMilli(event.getTimestamp()).toString(),
event.getSource()
);
}
public record AvroEventMessage(
String eventId, String type, String payload, String timestamp, String source) {}
}containerFactory = "avroKafkaListenerContainerFactory" connects this listener to the Avro factory. Without this attribute, Spring picks the @Primary factory (JSON), and the listener silently deserializes Avro binary as if it were JSON — producing garbage.
groupId = "labs-socket-avro" is a separate consumer group. This is critical. Using the same group as the JSON consumer (labs-socket-group) would cause both consumers to compete for partitions from two different topics under one group ID, breaking Kafka’s assignment model.
3.3 Cutover Sequence
Deploy labs-socket with the Avro consumer added. At this point, three things are happening simultaneously:
1. labs-api writes JSON to lab-events (avroEnabled=false)
2. labs-socket-group consumes JSON from lab-events
3. labs-socket-avro sits idle — lab-events.v2 has no messages yetStep 1: Enable dual-write in docker-compose.yml:
labs-api:
environment:
FEATURE_AVRO_ENABLED: "true"Redeploy labs-api. The producer now writes to both topics.
Step 2: Confirm both consumer groups are consuming and approaching lag=0:
./scripts/verify-parity.shExpected output when ready:
READY: Both consumer groups at lag=0.
Safe to disable JSON consumer and proceed to Phase 4 (cleanup).
Step 3: Disable the JSON @KafkaListener in labs-socket. This can be done by removing the annotation or setting the container to not auto-start:
// In the JSON consumer class — Phase 4 preparation
@KafkaListener(
topics = "${kafka.topic.lab-events:lab-events}",
groupId = "labs-socket-group",
autoStartup = "false" // disable — Avro consumer now handles all events
)
public void consume(LabEvent event, Acknowledgment ack) { ... }
autoStartup = "false" keeps the code in place (easy to re-enable for rollback) without the listener joining the consumer group.
Phase 4: Clean Up
After 48 hours of the Avro consumer running at lag=0 with no alerts:
- Remove the JSON
@KafkaListenerfromEventConsumer.java - Remove the
jsonKafkaTemplatesend path fromEventProducer.publish() - Remove the JSON producer factory bean from
KafkaProducerConfig - Remove the JSON consumer factory bean from
KafkaConsumerConfig - Delete the
lab-eventstopic (after confirming no other consumers) - Remove the
labs-socket-groupconsumer group offset — it no longer exists
# Confirm no active members in the old group before deletion
docker exec kafka-1 kafka-consumer-groups.sh \
--bootstrap-server localhost:9092 \
--describe --group labs-socket-group
# Delete topic — irreversible
docker exec kafka-1 kafka-topics.sh \
--bootstrap-server localhost:9092 \
--delete --topic lab-events
# Optional: rename lab-events.v2 to lab-events in topic config, application.yml, and re-registerRenaming is optional. If tooling expects lab-events, update the topic name in application.yml and Schema Registry subject. Otherwise, lab-events.v2 as a permanent name is fine.
Testing
The EmbeddedKafka Gap
@EmbeddedKafka starts an in-process Kafka broker. It does not start a Schema Registry.
A naive test that leaves SCHEMA_REGISTRY_URL=http://localhost:8081 will fail at test startup when Spring initializes the consumer factory — KafkaAvroDeserializer.configure() does not connect immediately, but the first poll() will fail trying to fetch the schema from a non-existent server.
// This does NOT work with @EmbeddedKafka without a running Schema Registry
@TestPropertySource(properties = "SCHEMA_REGISTRY_URL=http://localhost:8081")The Fix: mock:// URL
The kafka-schema-registry-client library (test dependency) intercepts any URL that starts with mock:// and routes all schema operations to an in-memory registry:
@SpringBootTest
@EmbeddedKafka(
partitions = 1,
topics = {"lab-events.v2"},
bootstrapServersProperty = "spring.kafka.bootstrap-servers"
)
@TestPropertySource(properties = {
"SCHEMA_REGISTRY_URL=mock://test", // activates in-memory registry
"kafka.topic.lab-events-v2=lab-events.v2"
})
@DirtiesContext
class AvroConsumerIntegrationTest {
@Autowired
private EmbeddedKafkaBroker embeddedKafka;
@MockBean
private SimpMessagingTemplate messaging;
@Test
void avroConsumer_broadcastsEvent_whenValidAvroMessageArrives() throws Exception {
com.labs.events.LabEvent event = com.labs.events.LabEvent.newBuilder()
.setEventId("test-001")
.setType("ORDER_CREATED")
.setPayload("{\"test\":true}")
.setTimestamp(Instant.now().toEpochMilli())
.setSource("integration-test")
.setCouponCode(null)
.build();
testAvroTemplate().send("lab-events.v2", event.getEventId(), event).get();
verify(messaging, timeout(5_000).atLeastOnce())
.convertAndSend(eq("/topic/events"), any(AvroEventConsumer.AvroEventMessage.class));
}
private KafkaTemplate<String, com.labs.events.LabEvent> testAvroTemplate() {
Map<String, Object> props = new HashMap<>();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, embeddedKafka.getBrokersAsString());
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
props.put(KafkaAvroSerializerConfig.SCHEMA_REGISTRY_URL_CONFIG, "mock://test");
return new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(props));
}
}The mock://test URL is a key. Any producer and consumer in the same JVM using the same key — test — share the same in-memory schema store. The test producer registers LabEvent on first send. The test consumer’s deserializer finds it in the same in-memory map. No HTTP. No external process.
This is the correct approach for all Avro consumer integration tests.
Parity Verification
After enabling dual-write in staging:
./scripts/verify-parity.sh
This script compares kafka-consumer-groups lag for both groups and the end offsets for both topics. If the Avro topic lags behind the JSON topic by more than a few messages after steady state, the dual-write is broken and should be investigated before proceeding to Phase 3.
Rollback Path
The migration is reversible at every phase because the feature flag controls activation:
Never delete a topic while consumers are assigned to it. Kafka assigns consumers to partitions — deleting the topic while the group is active orphans the consumer group and may cause deserialization errors in unrelated brokers if topic IDs are reused.
Common Pitfalls
Schema compatibility not configured
Schema Registry defaults to BACKWARD compatibility for new subjects. This means new schema versions must be readable by the previous version’s consumer. Adding a required field (without a default) breaks BACKWARD and Schema Registry will reject the registration. Always add new fields as optional unions with defaults.
To check the current compatibility setting:
curl -s http://localhost:8081/config/lab-event-value
Wrong SPECIFIC_AVRO_READER_CONFIG value
Without SPECIFIC_AVRO_READER_CONFIG=true, KafkaAvroDeserializer returns GenericRecord instead of the generated com.labs.events.LabEvent. The @KafkaListener method receives a GenericRecord, which cannot be cast to com.labs.events.LabEvent, and the consumer fails with a ClassCastException at runtime — not at startup. Easy to miss if the test only checks that the consumer received a message and not the type.
Missing containerFactory on @KafkaListener
Omitting containerFactory = "avroKafkaListenerContainerFactory" makes Spring use the @Primary factory (JSON). The listener will try to JSON-deserialize Avro binary messages and either fail or silently produce nulls. Always name the container factory explicitly on Avro listeners.
null union order matters
["null", "string"] is nullable-string with default null.["string", "null"] is non-null-string that also accepts null, with no valid null default.
Avro’s rule: the default value type must match the first type in a union. Getting this backwards causes org.apache.avro.SchemaParseException at code generation time — but only if the default is actually null and the non-null type is first.
Jackson version conflict
kafka-avro-serializer:7.6.0 pulls a specific Jackson version that may conflict with Spring Boot’s managed version. The <exclusion> for jackson-databind in pom.xml prevents the conflict. Without it, you may see NoSuchMethodError in Jackson at runtime.
Schema Registry uses INTERNAL listener, apps use PLAINTEXT
If labs-api or labs-socket cannot reach Schema Registry during the initial schema registration (first Avro send), the producer will throw io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException. Verify that SCHEMA_REGISTRY_URL points to http://schema-registry:8081 (inside Docker) and that Schema Registry has started healthy before the apps.
Week 4 Complete ✅
Key Takeaways
Never rename or delete a live topic during migration. The dual-write pattern creates lab-events.v2 alongside lab-events. Both topics exist and receive traffic simultaneously during the migration window. The old topic is only deleted after the new one has been stable for at least 48 hours.
The feature flag is the safety valve. FEATURE_AVRO_ENABLED=false is a zero-downtime rollback at any point in Phases 2 or 3. Do not remove it until Phase 4 cleanup is complete.
The @EmbeddedKafka gap is real. Avro serializers and deserializers contact Schema Registry during configure(). The mock://test URL from kafka-schema-registry-client provides an in-memory registry that works with @EmbeddedKafka without any external process. Every Avro consumer test should use it.
The complete source code is available on GitHub.
Support me through GitHub Sponsors.
Thank you for Reading !! See you in the next post.
Next
➡️ Week 5 · Day 29: Kafka Streams — KStream vs KTable
References
- 📘 Kafka: The Definitive Guide — Chapters 3 & 9
- 🌐 avro.apache.org/docs
- 🌐 docs.confluent.io/schema-registry
- Apache Avro 1.11 Specification
- Confluent Schema Registry API Reference
- Confluent Schema Registry Maven Repository
- Spring Kafka — Avro Serializers
- KafkaAvroSerializer Configuration
- Schema Registry Compatibility Types
