[
https://issues.apache.org/jira/browse/KAFKA-13681?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17497004#comment-17497004
]
Guozhang Wang commented on KAFKA-13681:
---------------------------------------
Thanks for reporting [~DrozD_0], I think this is indeed a place for
optimization and is related to
https://issues.apache.org/jira/browse/KAFKA-13676.
> Sink event duplicates for partition-stuck stream application
> ------------------------------------------------------------
>
> Key: KAFKA-13681
> URL: https://issues.apache.org/jira/browse/KAFKA-13681
> Project: Kafka
> Issue Type: Bug
> Components: streams
> Affects Versions: 2.8.1
> Reporter: Mikhail Dubrovin
> Priority: Major
> Attachments: fail_topology.txt
>
>
> Hello,
> We found the following unpredictable behavior of Kafka streams:
> {code:java}
> public void buildStreams(final BuilderHelper builder) {
> KTable<TableId, TableValue> table = builder.table();
> KTable<TableId, ArrayList<InternalWorkflowDTO>> workflowTable =
> workflowTable(builder);
> table
> .mapValues(value -> mappers.mainDTO(value))
> .leftJoin(workflowTable, mappers::joinWorkflows)
> .toStream()
> .map((key, value) -> KeyValue.pair(
> AggregateId.newBuilder().setId(value.getId()).build(),
> mappers.aggregateDTO(value)))
> .peek((k, v) -> logSinkRecord(v))
> .filter((id, dto) -> !isReprocessing)
> .to(...);
> }
> private static KTable<TableId, ArrayList<InternalWorkflowDTO>>
> workflowTable(BuilderHelper builderHelper) {
> return builderHelper.workflowTable()
> .groupBy((id, workflow) -> KeyValue.pair(
>
> TableId.newBuilder().setId(workflow.getTableId()).build(),
> mappers.mapWorkflow(workflow)),
> Grouped.with(...))
> .aggregate(ArrayList::new, (key, value, agg) -> {
> agg.add(value);
> return agg;
> }, (key, value, agg) -> {
> agg.remove(value);
> return agg;
> }, Materialized.with(...));
> } {code}
> it is a small part of our topology but it shows the error flow.
> *Data structure:*
> We have two many-partition topics: entity and workflow. Every topic is
> represented as KTable.
> *Data error that causes application shutdown:*
> Our final event(join the entity and workflow ktables) expects a not-null
> field in the entity but for some reason, it comes for one event. The whole
> aggregator fails in _mappers.aggregateDTO(value)_ of the _buildStreams_
> method
> We have a health check which restarts the aggregator if it fails.
> When incorrect data comes to one partition, the partition processing is stuck
> but other partitions are processed.
> It causes that at every restart, _workflowTable_ topology repeats
> .aggregate() add/remove flows and puts new List into the repartition topic.
> But offsets are not moved for processed partitions due to the aggregator's
> shutdown.
> _This behavior generates/sinks a lot of final entity duplicates at every
> restart because the flow is successful for data from a not-corrupted
> partition but offsets are not moved for them._
> And it also causes troubles if @EqualsAndHashCode is defined to use all
> fields to compare. At every restart, the topology tries to remove the old
> value(not existing after the first run) and adds a new value at the end of
> the list. The list grows after each restart(contains the same - new value
> values).
> I also attached the topology description. To visualize:
> [https://zz85.github.io/kafka-streams-viz/]
> *Current workaround:*
> To redefine @EqualsAndHashCode to use entities' ids only.
> *Not solved issue:*
> Sink events duplication at every restart.
> Thank you in advance!
--
This message was sent by Atlassian Jira
(v8.20.1#820001)
