DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
| Code Block |
|---|
// Producer Abortable Transaction
public class AbortableTransactionException extends ApiException {
public AbortableTransactionException(String message) {
super(message);
}
...
}
//Producer Retriable
public class ProducerRetriableTransactionException extends ApiException {
...
}
//Producer Refresh and Retriable
public ProducerRetriableTransactionException(String message) class ProducerRefreshRetriableTransactionException extends ApiException {
...
}
//Application-Recoverable
public class ApplicationRecoverableTransactionException extends super(message);ApiException {
}
...
}
//Producer Refresh and Retriable Invalid-Configuration
public class ProducerRefreshRetriableTransactionExceptionInvalidConfiguationTransactionException extends ApiException {
public ProducerRefreshRetriableTransactionException(String message) {
super(message);
}
...
}
//Application-Recoverable
public class ApplicationRecoverableTransactionException extends ApiException {
public ApplicationRecoverableTransactionException(String message) {
super(message);
}
...
}
// Invalid-Configuration
public class InvalidConfiguationTransactionException extends ApiException {
public InvalidConfiguationTransactionException(String message) {
super(message);
}
...
}
// Example of how exceptions will be extended by one of above exception types
public class InvalidProducerEpochException extends AbortableTransactionException {
private static final long serialVersionUID = 1L;
public InvalidProducerEpochException(String message) {
super(message);
}
} ...
}
|
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 (AbortableTransactionException e) {
// Abortable Exception: Handle Kafka exception by aborting transaction. producer.abortTransaction() should not throw abortable exception.
producer.abortTransaction();
resetToLastCommittedPositions(consumer);
}
} catch (InvalidConfiguationTransactionException e) {
// Fatal Error: The error is bubbled up to the application layer. The application can decide what to do
closeAll();
throw e;
} catch (KafkaException | ApplicationRecoverableTransactionException e) {
// Application Recoverable: The application must restart
closeAll();
initializeApplication();
}
}
} |
...