evolution-gaming / kafka-flow / 29419911541

83%

Build Type

push

Committed by web-flow

Commit Message

Derive transactional input topic and consumer generation from the flow (#842)

* Derive transactional config from the flow

The transactional Kafka snapshot persistence required hand-wiring two
values the flow already knows, with silent breakage when they mismatched:
the input topic was passed twice (once to the flow, again as
TransactionalConfig.inputTopic, unvalidated - a mismatch would commit
offsets for a topic the flow does not consume), and the driving consumer's
group metadata had to be passed explicitly, which forced allocating the
consumer before building the module and made it possible to wire the
metadata of some other consumer.

Both are now supplied by the flow itself: KafkaPersistenceModuleOf.make
takes the assigned TopicPartition instead of a bare Partition and
TransactionalConfig.inputTopic is gone, so the committed topic always
matches what the consumer reads; PartitionFlowOf.apply gains a
groupMetadata parameter that TopicFlow fills from the driving consumer,
and cachingTransactional loses its groupMetadata parameter - the module
is built like any other config-driven module and wired in as usual.

With both values out of the config, transactionalIdPrefix no longer has
a reason to encode the input topic: docs and examples now recommend the
applicationId (the prefix does not affect fencing - it is a readable
label and, on an ACL-secured cluster, the transactional.id prefix the
producer principal must be authorized for).

The per-assignment facts (topicPartition, assignedAt, groupMetadata) are
grouped into a single PartitionAssignment context, with the overridable
ScheduleCommit capability kept as a separate parameter;
KafkaPersistenceModuleOf reuses that type instead of defining its own, so
the transactional commit override happens in the capability parameter
rather than by rewriting the context.

The groupMetadata scaladoc states the live-read contract (pass it along
unevaluated - a memoized read would go stale on the next rebalance a... (continued)

Coverage Stats

2 of 2 new or added lines in 2 files covered. (100.0%)

269 existing lines in 68 files now uncovered.

1787 of 2132 relevant lines covered (83.82%)

1.3 hits per line