DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
| Table of Contents |
|---|
Status
Current state: Under Voting Complete
Discussion thread: here
JIRA: here
...
Exception/Error Names | Current handling | New Handling | |||
|---|---|---|---|---|---|
Producer API | Transaction API | Producer API | Transaction API | Comments | |
UnknownTopicOrPartitionException NotLeaderOrFollowerException | Retriable if the error is retriable. Otherwise abortable | Retriable if the error is retriable otherwise abortable | Refresh + Retriable | Refresh + Retriable | Both UnknownTopicOrPartitionException and NotLeaderOrFollowerException extends AuthorizationException InvalidMetadataException Current class hierarchy in code is: UnknownTopicOrPartitionException <InvalidMetadataException < RetriableException < ApiException
UnknownTopicOrPartitionException < InvalidMetadataException < < RetriableException < ApiException |
NotCoordinatorException CoordinatorNotAvailableException | Retriable | Retriable | Refresh + Retriable | Refresh + Retriable | |
...
| Code Block |
|---|
//Producer Refresh and Retriable
public abstract class RefreshRetriableException extends RetriableException {
...
}
//Application-Recoverable new
public abstract class ApplicationRecoverableException extends ApiException {
...
} |
We will extend below existing exceptions types to maintain class hierarchy
| Code Block |
|---|
// UnknownTopicOrPartitionException and NotLeaderOrFollowerException extends InvalidMetadataException
public class InvalidMetadataException extends RefreshRetriableException {
...
}
// TopicAuthorizationException and GroupAuthorizationException extends AuthorizationException
public class AuthorizationException extends InvalidConfigurationException {
...
} |
Client side code example
| Code Block |
|---|
public class TransactionalClientDemo {
private static final String CONSUMER_GROUP_ID = "my-group-id";
private static final String OUTPUT_TOPIC = "output";
private static final String INPUT_TOPIC = "input";
private static KafkaConsumer<String, String> consumer;
private static KafkaProducer<String, String> producer;
public static void main(String[] args) {
initializeApplication();
boolean isRunning = true;
// Continuously poll for records
while (isRunning) {
try {
try {
// Poll records from Kafka for a timeout of 60 seconds
ConsumerRecords<String, String> records = consumer.poll(ofSeconds(60));
// Process records to generate word count map
Map<String, Integer> wordCountMap = new HashMap<>();
for (ConsumerRecord<String, String> record : records) {
String[] words = record.value().split(" ");
for (String word : words) {
wordCountMap.merge(word, 1, Integer::sum);
}
}
// Begin transaction
producer.beginTransaction();
// Produce word count results to output topic
wordCountMap.forEach((key, value) ->
producer.send(new ProducerRecord<>(OUTPUT_TOPIC, key, value.toString())));
// Determine offsets to commit
Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = new HashMap<>();
for (TopicPartition partition : records.partitions()) {
List<ConsumerRecord<String, String>> partitionedRecords = records.records(partition);
long offset = partitionedRecords.get(partitionedRecords.size() - 1).offset();
offsetsToCommit.put(partition, new OffsetAndMetadata(offset + 1));
}
// Send offsets to transaction for atomic commit
producer.sendOffsetsToTransaction(offsetsToCommit, CONSUMER_GROUP_ID);
// Commit transaction
producer.commitTransaction();
} catch (TransactionAbortableException e) {
// Abortable Exception: Handle Kafka exception by aborting transaction. producer.abortTransaction() should not throw abortable exception.
producer.abortTransaction();
resetToLastCommittedPositions(consumer);
}
} catch (InvalidConfigurationException e) {
// Fatal Error: The error is bubbled up to the application layer. The application can decide what to do
closeAll();
throw e;
} catch (KafkaException | ApplicationRecoverableException e) {
// Application Recoverable: The application must restart
closeAll();
initializeApplication();
}
}
} |
...
Currently, the transactional producer.send returns retriable exception types, such as TimeoutException , which poses a risk of duplicates in Kafka. In this KIP, we will update the transactional producer.send path such that all retriable exceptions will be translated to TransactionAbortableException in transaction producer code path. Older clients that are using the transactional producer and handling TimeoutException by retrying the produce operation can update to handle TransactionAbortableException .
...