Day 5: Engagement Consumer Development

• streamsocial-engagement-consumer: the first consumer, reading content-interactions (Day 3) for likes, comments, shares

• @KafkaListener with Spring’s JsonDeserializer<ContentInteractionEvent>, symmetric with Day 4’s producer-side serialization choices

• A second dashboard panel, live, alongside the first — nothing about Day 4’s panel changes

The error-handling class this lesson was originally scoped around, SeekToCurrentErrorHandler, doesn’t exist in current Spring Kafka. It was deprecated in version 2.8 and removed outright in 3.x. That’s the kind of thing that happens constantly in a fast-moving ecosystem, and pretending otherwise would mean shipping a lesson that doesn’t compile against anything you’d actually install today. The replacement, DefaultErrorHandler, does the same job the deprecated class did: on a listener failure, it seeks the failed record — and everything after it in that poll batch — back to be redelivered by the broker, retries against a configurable backoff, and only after retries are exhausted hands the record to a recoverer instead of retrying forever or crashing the consumer thread.

A Kafka consumer doesn’t get pushed messages — it polls. poll() asks the broker for the next batch of records sitting after the consumer’s current offset on whatever partitions it’s assigned, gets a batch back, and the application processes that batch before calling poll() again. @KafkaListener hides this loop entirely: Spring’s listener container runs it on a background thread and calls your annotated method once per record. What you’re writing is the body of that loop, not the loop itself.

Day 4’s producer configured JsonSerializer.ADD_TYPE_INFO_HEADERS = false — no type metadata riding along on every record, because every message on a given topic is one known type and saying so on every single message is waste. That decision has a mirror-image requirement on the consuming side: JsonDeserializer needs to be told the target type directly, new JsonDeserializer<>(ContentInteractionEvent.class, false), rather than expecting a header that was never sent. Producer and consumer configuration aren’t independent — a serialization decision made in one module is a contract the other module has to honor, and getting that pairing wrong is a silent, confusing failure mode (deserialization exceptions with no obvious cause) rather than a loud one. Today’s own EngagementTestDataGenerator follows the exact same wire format for this reason — it’s standing in for a real content-interactions producer this course hasn’t built yet, and doing so faithfully matters.

Successfully deserializing a record proves the bytes were well-formed JSON matching the expected shape. It proves nothing about whether the values make sense. This listener runs every deserialized ContentInteractionEvent through the same jakarta.validation.Validator Day 1 used — a blank contentId that somehow made it onto the topic is caught here, thrown as an exception, and handled by the same retry-then-recover path as any other processing failure. Bean Validation isn’t a REST-layer-only concern; it’s useful anywhere a payload crosses a trust boundary, and a Kafka topic populated by services you don’t fully control is exactly that.

Today’s failure hook — a deliberately poisoned contentId of post-BOOM — exists to make the failure path observable, not because that’s a real production scenario. What matters is what happens when it fires: the record gets retried twice, one second apart, and if it still fails, RecoveryTracker logs it as STRUCTURED_ERROR and counts it — instead of the consumer thread dying (which would stop every other partition assigned to it from being processed too) or the record being silently dropped. “Graceful” here specifically means: bounded retries, then a loud, greppable record of what got given up on. It does not yet mean routing the record somewhere it can be reprocessed later — that’s a dead letter topic, and that’s a later lesson’s job, once poison-pill handling gets the dedicated treatment it deserves.

ContentInteractionsFeedListener follows the exact shape UserActionsFeedListener established Day 4 — its own consumer group (dashboard-live-feed-content-interactions, deliberately distinct from both engagement-consumer‘s group and the user-actions panel’s own group), broadcasting through the same EventStreamBroadcaster to a second SSE stream. Nothing about the user-actions panel changes; the two feeds run side by side, each backed by its own independent, observational consumer. Run EngagementTestDataGenerator and the same 14 events show up in two places at once — streamsocial-engagement-consumer‘s own log (processing them for real, with validation and error handling) and the dashboard’s browser view (just watching, no business logic at all) — which is itself a small, live demonstration of what a consumer group actually buys you: two entirely independent readers of the same topic, neither aware of the other, both correct.

https://github.com/sysdr/streamsocial-java/tree/main/day05-v2-streamsocial-source/streamsocial

cd streamsocial-infra && ./scripts/start.sh && cd ..
mvn org.codehaus.mojo:exec-maven-plugin:3.3.0:java -pl streamsocial-common \
  -Dexec.mainClass=com.streamsocial.common.demo.TopicBootstrapDemo \
  -Dstreamsocial.topics.user-actions-partitions=12 \
  -Dstreamsocial.topics.content-interactions-partitions=6

mvn -pl streamsocial-dashboard -am spring-boot:run

Open http://localhost:8080 — two panels now, user-actions and content-interactions, both at 0.

In another terminal:

mvn -pl streamsocial-engagement-consumer -am spring-boot:run

Expected: Spring Boot startup log, ending quietly — no web server line, since this module has none.

mvn org.codehaus.mojo:exec-maven-plugin:3.3.0:java -pl streamsocial-engagement-consumer \
  -Dexec.mainClass=com.streamsocial.consumer.demo.EngagementTestDataGenerator

Expected in the consumer’s terminal: 14 lines like:

STRUCTURED_EVENT event=engagement-processed interactionType=CONTENT_LIKED userId=user-0 contentId=post-0 partition=1 offset=0

Then, a couple seconds later:

STRUCTURED_ERROR event=engagement-processing-recovered topic=content-interactions partition=... offset=... key=post-BOOM reason=simulated processing failure for post-BOOM

Expected on the dashboard, at roughly the same time: the content-interactions panel’s counter climbs to 14 (not 15 — the poisoned event never succeeds, so it never gets broadcast either, the same way it never lands in the consumer’s processed list) and 14 rows appear.

mvn -pl streamsocial-engagement-consumer -am verify
mvn -pl streamsocial-dashboard -am verify

Expected: all green. EngagementEventListenerIT‘s poisonedEventIsRetriedThenRecoveredNotLostSilently takes noticeably longer than the other tests — it’s waiting out the real retry backoff, not asserting instantly. ContentInteractionsFeedBroadcastIT proves the new panel’s data path the same way Day 4’s UserActionsFeedBroadcastIT did for the first one.

EngagementEventListenerIT passes — both the normal-processing case and the poisoned-record recovery case, against a real broker, with the real listener bean. Run the test-data generator against a live consumer and dashboard together, and find STRUCTURED_EVENT/STRUCTURED_ERROR lines in the consumer’s log at the same moment matching rows appear on the dashboard’s second panel.

Add a second @KafkaListener method to a new class, EngagementMetricsListener, that listens to content-interactions but with a different group.id. Have it maintain a simple in-memory count per InteractionType.

Implementation checklist:

1. New listener class, new @KafkaListener(topics = "content-interactions", groupId = "engagement-metrics") with its own ConsumerFactory/container factory bean, following KafkaConsumerConfig‘s existing pattern.

2. A ConcurrentHashMap<InteractionType, Long> (or similar) tallying counts, with a getter for tests.

3. A test that publishes a known mix of interaction types and asserts the final counts match.

4. Run both listeners at once and confirm both process every message — think about why a different group.id makes that true, tying back to what today’s dashboard panels already demonstrated.

• Two consumers in two different groups both get their own copy of every record — consumer groups are what make records get split across consumers, not consumers in different groups compete for the same records. The dashboard’s two independent panels reading the same topics as the business consumers are already a live example of exactly this.

• Your metrics listener doesn’t need the DefaultErrorHandler/retry setup this lesson’s main listener has — it’s fine for homework purposes to let a bad record just log and move on, since the goal here is understanding group isolation, not re-solving today’s error handling.

• If your counts come out lower than expected, double-check you created a new, distinctly-named container factory bean for the new group ID rather than accidentally reusing engagementListenerContainerFactory, which is already bound to group.id=engagement-consumer at the factory level.