Skip to content

KAFKA-14095: Improve handling of sync offset failures in MirrorMaker #12432

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

Merged
mimaison merged 2 commits into from
Aug 1, 2022
Merged

KAFKA-14095: Improve handling of sync offset failures in MirrorMaker #12432

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
91f4446
KAFKA-14095: Improve handling of sync offset failures in MirrorMaker
mimaison Jul 22, 2022
551a486
Address reviews
mimaison Jul 29, 2022
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
3 changes: 3 additions & 0 deletions .../main/java/org/apache/kafka/clients/admin/internals/AlterConsumerGroupOffsetsHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -179,6 +179,9 @@ private void handleError(
case INVALID_GROUP_ID:
case INVALID_COMMIT_OFFSET_SIZE:
case GROUP_AUTHORIZATION_FAILED:
// Member level errors.
case UNKNOWN_MEMBER_ID:
case FENCED_INSTANCE_ID:

@C0urante C0urante Jul 29, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

When would we encounter this error? My understanding is that it'll only be returned when the broker receives a request with a group instance ID defined, and IIUC it's not possible to define one with the admin API we expose right now.

@mimaison mimaison Jul 29, 2022

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

You're right, currently this error code is not expected as the admin client does not set the group instance ID. I've removed it.

log.debug("OffsetCommit request for group id {} failed due to error {}.",
groupId.idValue, error);
partitionResults.put(topicPartition, error);
Expand Down
13 changes: 12 additions & 1 deletion connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorCheckpointTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,9 +17,11 @@
package org.apache.kafka.connect.mirror;

import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.AlterConsumerGroupOffsetsResult;
import org.apache.kafka.clients.admin.ConsumerGroupDescription;
import org.apache.kafka.common.ConsumerGroupState;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.errors.UnknownMemberIdException;
import org.apache.kafka.connect.source.SourceTask;
import org.apache.kafka.connect.source.SourceRecord;
import org.apache.kafka.connect.data.Schema;
Expand Down Expand Up @@ -306,7 +308,16 @@ Map<String, Map<TopicPartition, OffsetAndMetadata>> syncGroupOffset() {

void syncGroupOffset(String consumerGroupId, Map<TopicPartition, OffsetAndMetadata> offsetToSync) {
if (targetAdminClient != null) {
targetAdminClient.alterConsumerGroupOffsets(consumerGroupId, offsetToSync);
AlterConsumerGroupOffsetsResult result = targetAdminClient.alterConsumerGroupOffsets(consumerGroupId, offsetToSync);

@C0urante C0urante Jul 29, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Nit: would it also be useful to have a log line here indicating that we're attempting to sync offsets?

@mimaison mimaison Aug 1, 2022

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

The Scheduler prints an info line each time it runs this:

INFO [MirrorCheckpointConnector|task-0] sync idle consumer group offset from source to target took 0 ms (org.apache.kafka.connect.mirror.Scheduler:95)

So I'll leave it like this for now.

result.all().whenComplete((v, throwable) -> {
if (throwable != null) {
if (throwable.getCause() instanceof UnknownMemberIdException) {

@C0urante C0urante Jul 29, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

How stable is the API for the exceptions we'll see here? Can we be reasonably certain that the exception we want to examine will always be wrapped?

@mimaison mimaison Jul 29, 2022

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

The outer exception depends how the future is completed but I think the actual error should always be wrapped.

@C0urante C0urante Jul 29, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

I took a closer look at the KafkaFuture Javadocs.

The docs for KafkaFuture::whenComplete state (emphasis mine):

Returns a new KafkaFuture with the same result or exception as this future, that executes the given action when this future completes. When this future is done, the given action is invoked with the result (or null if none) and the exception (or null if none) of this future as arguments."

Given this, we know that whenComplete receives the same exception that would be thrown by, e.g., KafkaFuture::get, if the action failed.

The method signatures for KafkaFuture declare that they throw ExecutionException if the action fails, which always wraps the cause of failure.

So, I'm convinced that we can expect this wrapping 👍

log.warn("Unable to sync offsets for consumer group {}. This is likely caused by consumers currently using this group in the target cluster.", consumerGroupId);
} else {
log.error("Unable to sync offsets for consumer group {}.", consumerGroupId, throwable);
}
}
});
log.trace("sync-ed the offset for consumer group: {} with {} number of offset entries",

@C0urante C0urante Jul 29, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Should we update this log message since it's not guaranteed that the sync will have completed by this point, or that it will even complete successfully at all?

Or alternatively, could we keep the message as-is, but move it into an else block in the callback we pass to whenComplete?

@mimaison mimaison Jul 29, 2022

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

I've moved it in an else block above and tweaked the message slightly.

consumerGroupId, offsetToSync.size());
}
Expand Down