Skip to content

KAFKA-15425: Fail fast in Admin::listOffsets when topic (but not partition) metadata is not found #14314

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
jolshan merged 3 commits into from
Sep 7, 2023
Merged

KAFKA-15425: Fail fast in Admin::listOffsets when topic (but not partition) metadata is not found #14314

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
246acd5
KAFKA-15425: Fail fast in Admin::listOffsets when topic (but not part…
C0urante Aug 30, 2023
8954742
Parameterize KafkaAdminClientTest::testListOffsetsMetadataNonRetriabl…
C0urante Sep 7, 2023
d0eb999
Add more test cases to KafkaAdminClientTest::testListOffsetsMetadataN…
C0urante Sep 7, 2023
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
2 changes: 1 addition & 1 deletion clients/src/main/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,7 +61,7 @@ public ListOffsetsHandler(
this.offsetTimestampsByPartition = offsetTimestampsByPartition;
this.options = options;
this.log = logContext.logger(ListOffsetsHandler.class);
this.lookupStrategy = new PartitionLeaderStrategy(logContext);
this.lookupStrategy = new PartitionLeaderStrategy(logContext, false);
}

@Override
Expand Down
14 changes: 14 additions & 0 deletions clients/src/main/java/org/apache/kafka/clients/admin/internals/PartitionLeaderStrategy.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,9 +42,15 @@ public class PartitionLeaderStrategy implements AdminApiLookupStrategy<TopicPart
};

private final Logger log;
private final boolean tolerateUnknownTopics;

public PartitionLeaderStrategy(LogContext logContext) {
this(logContext, true);
}

public PartitionLeaderStrategy(LogContext logContext, boolean tolerateUnknownTopics) {
this.log = logContext.logger(PartitionLeaderStrategy.class);
this.tolerateUnknownTopics = tolerateUnknownTopics;
}

@Override
Expand All @@ -64,6 +70,7 @@ public MetadataRequest.Builder buildRequest(Set<TopicPartition> partitions) {
return new MetadataRequest.Builder(request);
}

@SuppressWarnings("fallthrough")
private void handleTopicError(
String topic,
Errors topicError,
Expand All @@ -72,6 +79,12 @@ private void handleTopicError(
) {
switch (topicError) {
case UNKNOWN_TOPIC_OR_PARTITION:
if (!tolerateUnknownTopics) {

@jolshan jolshan Sep 5, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

the operation would be retried if any metadata error was reported for any individual topic partition, even if an error was also reported for the entire topic. With this change, the operation always fails if an error is reported for the entire topic, even if an error is also reported for one or more individual topic partitions.

This change only changes the behavior for unknown topic or partition errors right? Leader/broker not available continue to be retriable, topic auth, invalid topic, and other errors continue to fail the request?

@jolshan jolshan Sep 6, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

am not aware of any cases where brokers might return both topic- and topic partition-level errors for a metadata request, and if there are none, then this change should be safe.

Asking because my understanding is the only case where we would be concerned about topic and partition level errors is when the topic level error is unknown topic or partition but somehow the partitions for that topic have a different error?

@C0urante C0urante Sep 7, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

This change only changes the behavior for unknown topic or partition errors right? Leader/broker not available continue to be retriable, topic auth, invalid topic, and other errors continue to fail the request?

Exactly, and we have unit testing coverage to demonstrate this.

Asking because my understanding is the only case where we would be concerned about topic and partition level errors is when the topic level error is unknown topic or partition but somehow the partitions for that topic have a different error?

This is correct, and I'm optimistic that no such case exists, but wanted to note the potential change in behavior just to be safe.

log.error("Received unknown topic error for topic {}", topic, topicError.exception());
failAllPartitionsForTopic(topic, requestPartitions, failed, tp -> topicError.exception(
"Failed to fetch metadata for partition " + tp + " because metadata for topic `" + topic + "` could not be found"));
break;
}
case LEADER_NOT_AVAILABLE:
case BROKER_NOT_AVAILABLE:
log.debug("Metadata request for topic {} returned topic-level error {}. Will retry",
Expand Down Expand Up @@ -124,6 +137,7 @@ private void handlePartitionError(
case LEADER_NOT_AVAILABLE:
case BROKER_NOT_AVAILABLE:
case KAFKA_STORAGE_ERROR:
case UNKNOWN_TOPIC_OR_PARTITION:
log.debug("Metadata request for partition {} returned partition-level error {}. Will retry",
topicPartition, partitionError);
break;
Expand Down
61 changes: 51 additions & 10 deletions clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -212,6 +212,8 @@
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
import org.junit.jupiter.params.provider.ValueSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand Down Expand Up @@ -557,12 +559,16 @@ private static FindCoordinatorResponse prepareBatchedFindCoordinatorResponse(Err
}

private static MetadataResponse prepareMetadataResponse(Cluster cluster, Errors error) {
return prepareMetadataResponse(cluster, error, error);
}

private static MetadataResponse prepareMetadataResponse(Cluster cluster, Errors topicError, Errors partitionError) {
List<MetadataResponseTopic> metadata = new ArrayList<>();
for (String topic : cluster.topics()) {
List<MetadataResponsePartition> pms = new ArrayList<>();
for (PartitionInfo pInfo : cluster.availablePartitionsForTopic(topic)) {
MetadataResponsePartition pm = new MetadataResponsePartition()
.setErrorCode(error.code())
.setErrorCode(partitionError.code())
.setPartitionIndex(pInfo.partition())
.setLeaderId(pInfo.leader().id())
.setLeaderEpoch(234)
Expand All @@ -572,7 +578,7 @@ private static MetadataResponse prepareMetadataResponse(Cluster cluster, Errors
pms.add(pm);
}
MetadataResponseTopic tm = new MetadataResponseTopic()
.setErrorCode(error.code())
.setErrorCode(topicError.code())
.setName(topic)
.setIsInternal(false)
.setPartitions(pms);
Expand Down Expand Up @@ -5441,7 +5447,6 @@ public void testDescribeMetadataQuorumFailure() {

@Test
public void testListOffsetsMetadataRetriableErrors() throws Exception {

Node node0 = new Node(0, "localhost", 8120);
Node node1 = new Node(1, "localhost", 8121);
List<Node> nodes = Arrays.asList(node0, node1);
Expand All @@ -5464,7 +5469,8 @@ public void testListOffsetsMetadataRetriableErrors() throws Exception {
env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());

env.kafkaClient().prepareResponse(prepareMetadataResponse(cluster, Errors.LEADER_NOT_AVAILABLE));
env.kafkaClient().prepareResponse(prepareMetadataResponse(cluster, Errors.UNKNOWN_TOPIC_OR_PARTITION));
// We retry when a partition of a topic (but not the topic itself) is unknown
env.kafkaClient().prepareResponse(prepareMetadataResponse(cluster, Errors.NONE, Errors.UNKNOWN_TOPIC_OR_PARTITION));
env.kafkaClient().prepareResponse(prepareMetadataResponse(cluster, Errors.NONE));

// listoffsets response from broker 0
Expand Down Expand Up @@ -5615,9 +5621,13 @@ public void testListOffsetsWithLeaderChange() throws Exception {
}
}

@Test
public void testListOffsetsMetadataNonRetriableErrors() throws Exception {

@ParameterizedTest
@MethodSource("listOffsetsMetadataNonRetriableErrors")
public void testListOffsetsMetadataNonRetriableErrors(
Errors topicMetadataError,
Errors partitionMetadataError,
Class<? extends Throwable> expectedFailure
) throws Exception {
Node node0 = new Node(0, "localhost", 8120);
Node node1 = new Node(1, "localhost", 8121);
List<Node> nodes = Arrays.asList(node0, node1);
Expand All @@ -5633,18 +5643,49 @@ public void testListOffsetsMetadataNonRetriableErrors() throws Exception {
node0);

final TopicPartition tp1 = new TopicPartition("foo", 0);
final MetadataResponse preparedResponse = prepareMetadataResponse(
cluster, topicMetadataError, partitionMetadataError
);

try (AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(cluster)) {
env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());

env.kafkaClient().prepareResponse(prepareMetadataResponse(cluster, Errors.TOPIC_AUTHORIZATION_FAILED));
env.kafkaClient().prepareResponse(preparedResponse);

Map<TopicPartition, OffsetSpec> partitions = new HashMap<>();
partitions.put(tp1, OffsetSpec.latest());
ListOffsetsResult result = env.adminClient().listOffsets(partitions);

TestUtils.assertFutureError(result.all(), TopicAuthorizationException.class);
}
TestUtils.assertFutureError(result.all(), expectedFailure);
}
}

private static Stream<Arguments> listOffsetsMetadataNonRetriableErrors() {
return Stream.of(
Arguments.of(
Errors.TOPIC_AUTHORIZATION_FAILED,
Errors.TOPIC_AUTHORIZATION_FAILED,
TopicAuthorizationException.class
),
Arguments.of(
// We fail fast when the entire topic is unknown...
Errors.UNKNOWN_TOPIC_OR_PARTITION,
Errors.NONE,

@jolshan jolshan Sep 7, 2023

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

For my understanding, if the topic also had unknown topic or partition as the partition error message, we would fail fast there too.

With this change, the operation always fails if an error is reported for the entire topic, even if an error is also reported for one or more individual topic partitions.

Would we want to add this test case too just so the behavior is also shown in tests?

@C0urante C0urante Sep 7, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Sure, done. I've tried to clarify in the comments that these are unusual cases we're testing for lest anyone think that this is normal broker behavior.

UnknownTopicOrPartitionException.class
),
Arguments.of(
// ... even if a partition in the topic is also somehow reported as unknown...
Errors.UNKNOWN_TOPIC_OR_PARTITION,
Errors.UNKNOWN_TOPIC_OR_PARTITION,
UnknownTopicOrPartitionException.class
),
Arguments.of(
// ... or a partition in the topic has a different, otherwise-retriable error
Errors.UNKNOWN_TOPIC_OR_PARTITION,
Errors.LEADER_NOT_AVAILABLE,
UnknownTopicOrPartitionException.class
)
);
}

@Test
Expand Down