Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

KAFKA-9113: Unit test for RecordCollector #5

Merged
merged 4 commits into from
Jan 22, 2020
Merged

KAFKA-9113: Unit test for RecordCollector #5

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
955702c
fix record collector unit tests
guozhangwang Jan 12, 2020
305417e
add one more test
guozhangwang Jan 13, 2020
ea37738
Merge branch 'k9113-base' of https://github.com/guozhangwang/kafka in…
guozhangwang Jan 16, 2020
327d6a9
Boyang's comments
guozhangwang Jan 19, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 10 additions & 5 deletions streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -132,6 +132,8 @@ private void maybeBeginTxn() {
producer.beginTransaction();
} catch (final ProducerFencedException error) {
throw new TaskMigratedException(taskId, "Producer get fenced trying to begin a new transaction", error);
} catch (final KafkaException error) {
throw new StreamsException("Producer encounter unexpected error trying to begin a new transaction", error);
}
transactionInFlight = true;
}
Expand All @@ -150,16 +152,18 @@ private void maybeAbortTxn() {
*/

// can be ignored: transaction got already aborted by brokers/transactional-coordinator if this happens
} catch (final KafkaException error) {
throw new StreamsException("Producer encounter unexpected error trying to abort the transaction", error);
}
transactionInFlight = false;
}
}

public void commit(final Map<TopicPartition, OffsetAndMetadata> offsets) {
if (eosEnabled) {
try {
maybeBeginTxn();
maybeBeginTxn();

try {
producer.sendOffsetsToTransaction(offsets, applicationId);
producer.commitTransaction();
transactionInFlight = false;
Expand Down Expand Up @@ -247,6 +251,7 @@ public <K, V> void send(final String topic,
final StreamPartitioner<? super K, ? super V> partitioner) {
final Integer partition;

// TODO K9113: we need to decide how to handle exceptions from partitionsFor
if (partitioner != null) {
final List<PartitionInfo> partitions = producer.partitionsFor(topic);
if (partitions.size() > 0) {
Expand All @@ -273,9 +278,9 @@ public <K, V> void send(final String topic,
final Serializer<V> valueSerializer) {
checkForException();

try {
maybeBeginTxn();
maybeBeginTxn();

try {
final byte[] keyBytes = keySerializer.serialize(topic, headers, key);
final byte[] valBytes = valueSerializer.serialize(topic, headers, value);

Expand All @@ -301,7 +306,7 @@ public <K, V> void send(final String topic,
if (isRecoverable(uncaughtException)) {
// producer.send() call may throw a KafkaException which wraps a FencedException,
// in this case we should throw its wrapped inner cause so that it can be captured and re-wrapped as TaskMigrationException
throw new TaskMigratedException(taskId, "Producer cannot send records anymore since it got fenced", uncaughtException);
throw new TaskMigratedException(taskId, "Producer cannot send records anymore since it got fenced", uncaughtException.getCause());
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

q: this assumes the uncaught exception is always wrapped, is that the case?

Copy link
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The isRecoverable will only return true if 1) itself is a KafkaException, and 2) it has a wrapped exception of ProducerFencedException or UnknownProducerIdException. So here we are certain that it has a wrapped exception and we are only interested in that exception.

} else {
final String errorMessage = String.format(SEND_EXCEPTION_MESSAGE, topic, taskId, uncaughtException.toString());
throw new StreamsException(errorMessage, uncaughtException);
Expand Down
Loading