[
https://issues.apache.org/jira/browse/KAFKA-7108?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16524716#comment-16524716
]
Ted Yu edited comment on KAFKA-7108 at 6/27/18 7:43 AM:
--------------------------------------------------------
In StreamTask#commitOffsets :
{code}
if (eosEnabled) {
producer.sendOffsetsToTransaction(consumedOffsetsAndMetadata, applicationId);
{code}
Should we add try-catch block around the above call and check whether
e.getCause() instanceof NotEnoughReplicasException ?
How many times should the code retry sending offsets ?
was (Author: yuzhih...@gmail.com):
{code}
if (eosEnabled) {
producer.sendOffsetsToTransaction(consumedOffsetsAndMetadata, applicationId);
{code}
Should we add try-catch block around the above call and check whether
e.getCause() instanceof NotEnoughReplicasException ?
How many times should the code retry sending offsets ?
> "Exactly-once" stream breaks production exception handler contract
> ------------------------------------------------------------------
>
> Key: KAFKA-7108
> URL: https://issues.apache.org/jira/browse/KAFKA-7108
> Project: Kafka
> Issue Type: Bug
> Components: streams
> Affects Versions: 1.1.0
> Reporter: Anna O
> Priority: Critical
> Labels: exactly-once
>
> I have a stream configured with "default.production.exception.handler" that
> is supposed to log the error and continue. When I set "processing.guarantee"
> to "exactly_once" it appeared that retryable NotEnoughReplicasException that
> passed the production exception handler was rethrown by the
> TransactionManager wrapped with KafkaException and terminated the stream
> thread:
> _org.apache.kafka.common.KafkaException: Cannot execute transactional method
> because we are in an error stateat
> org.apache.kafka.clients.producer.internals.TransactionManager.maybeFailWithError(TransactionManager.java:784)
> ~[kafka-clients-1.1.0.jar:?]_
> _at
> org.apache.kafka.clients.producer.internals.TransactionManager.sendOffsetsToTransaction(TransactionManager.java:250)
> ~[kafka-clients-1.1.0.jar:?]_
> _at
> org.apache.kafka.clients.producer.KafkaProducer.sendOffsetsToTransaction(KafkaProducer.java:617)
> ~[kafka-clients-1.1.0.jar:?]_
> _at
> org.apache.kafka.streams.processor.internals.StreamTask.commitOffsets(StreamTask.java:357)
> ~[kafka-streams-1.1.0.jar:?]_
> _at
> org.apache.kafka.streams.processor.internals.StreamTask.access$000(StreamTask.java:53)
> ~[kafka-streams-1.1.0.jar:?]_
> _at
> org.apache.kafka.streams.processor.internals.StreamTask$1.run(StreamTask.java:316)
> ~[kafka-streams-1.1.0.jar:?]_
> _at
> org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:208)
> ~[kafka-streams-1.1.0.jar:?]_
> _at
> org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:307)
> ~[kafka-streams-1.1.0.jar:?]_
> _at
> org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:297)
> ~[kafka-streams-1.1.0.jar:?]_
> _at
> org.apache.kafka.streams.processor.internals.AssignedTasks$1.apply(AssignedTasks.java:67)
> ~[kafka-streams-1.1.0.jar:?]_
> _at
> org.apache.kafka.streams.processor.internals.AssignedTasks.applyToRunningTasks(AssignedTasks.java:357)
> [kafka-streams-1.1.0.jar:?]_
> _at
> org.apache.kafka.streams.processor.internals.AssignedTasks.commit(AssignedTasks.java:347)
> [kafka-streams-1.1.0.jar:?]_
> _at
> org.apache.kafka.streams.processor.internals.TaskManager.commitAll(TaskManager.java:403)
> [kafka-streams-1.1.0.jar:?]_
> _at
> org.apache.kafka.streams.processor.internals.StreamThread.maybeCommit(StreamThread.java:994)
> [kafka-streams-1.1.0.jar:?]_
> _at
> org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:811)
> [kafka-streams-1.1.0.jar:?]_
> _at
> org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:750)
> [kafka-streams-1.1.0.jar:?]_
> _at
> org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:720)
> [kafka-streams-1.1.0.jar:?]_
> _Caused by: org.apache.kafka.common.errors.NotEnoughReplicasException:
> Messages are rejected since there are fewer in-sync replicas than required._
>
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
