Skip to content

KAFKA-10223; Use NOT_LEADER_OR_FOLLOWER instead of non-retriable REPLICA_NOT_AVAILABLE for consumers #8979

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
rajinisivaram merged 5 commits into from
Jul 17, 2020
Merged

KAFKA-10223; Use NOT_LEADER_OR_FOLLOWER instead of non-retriable REPLICA_NOT_AVAILABLE for consumers #8979

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
116b7b0
KAFKA-10223; Replace REPLICA_NOT_AVAILABLE with NOT_LEADER_OR_FOLLOWER
rajinisivaram Jul 9, 2020
8bcda5e
Address review comments, remove some more usages of ReplicaNotAvailable
rajinisivaram Jul 14, 2020
eedd6e2
Address review comments
rajinisivaram Jul 16, 2020
991b61d
Address review comments
rajinisivaram Jul 17, 2020
65d4f62
Address review comment
rajinisivaram Jul 17, 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
4 changes: 3 additions & 1 deletion clients/src/main/java/org/apache/kafka/clients/admin/AlterReplicaLogDirsResult.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
import org.apache.kafka.common.errors.InvalidTopicException;
import org.apache.kafka.common.errors.KafkaStorageException;
import org.apache.kafka.common.errors.LogDirNotFoundException;
import org.apache.kafka.common.errors.NotLeaderOrFollowerException;
import org.apache.kafka.common.errors.ReplicaNotAvailableException;
import org.apache.kafka.common.errors.UnknownServerException;

Expand Down Expand Up @@ -59,7 +60,8 @@ public class AlterReplicaLogDirsResult {
* <li>{@link ClusterAuthorizationException}: Authorization failed. (CLUSTER_AUTHORIZATION_FAILED, 31)</li>
* <li>{@link InvalidTopicException}: The specified topic name is too long. (INVALID_TOPIC_EXCEPTION, 17)</li>
* <li>{@link LogDirNotFoundException}: The specified log directory is not found in the broker. (LOG_DIR_NOT_FOUND, 57)</li>
* <li>{@link ReplicaNotAvailableException}: The replica does not exist on the broker. (REPLICA_NOT_AVAILABLE, 9)</li>
* <li>{@link ReplicaNotAvailableException}: The replica does not exist on the broker with inter-broker protocol version < 2.7. (REPLICA_NOT_AVAILABLE, 9)</li>
* <li>{@link NotLeaderOrFollowerException}: The replica does not exist on the broker with inter-broker protocol version >= 2.7. (NOT_LEADER_OR_FOLLOWER, 6)</li>
* <li>{@link KafkaStorageException}: Disk error occurred. (KAFKA_STORAGE_ERROR, 56)</li>
* <li>{@link UnknownServerException}: Unknown. (UNKNOWN_SERVER_ERROR, -1)</li>
* </ul>
Expand Down
4 changes: 2 additions & 2 deletions clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1043,7 +1043,7 @@ private void handleListOffsetResponse(Map<TopicPartition, ListOffsetRequest.Part
log.debug("Cannot search by timestamp for partition {} because the message format version " +
"is before 0.10.0", topicPartition);
break;
case NOT_LEADER_FOR_PARTITION:
case NOT_LEADER_OR_FOLLOWER:
case REPLICA_NOT_AVAILABLE:
case KAFKA_STORAGE_ERROR:
case OFFSET_NOT_AVAILABLE:
Expand Down Expand Up @@ -1274,7 +1274,7 @@ private CompletedFetch initializeCompletedFetch(CompletedFetch nextCompletedFetc
}

nextCompletedFetch.initialized = true;
} else if (error == Errors.NOT_LEADER_FOR_PARTITION ||
} else if (error == Errors.NOT_LEADER_OR_FOLLOWER ||
error == Errors.REPLICA_NOT_AVAILABLE ||
error == Errors.KAFKA_STORAGE_ERROR ||
error == Errors.FENCED_LEADER_EPOCH ||
Expand Down
2 changes: 1 addition & 1 deletion ...rc/main/java/org/apache/kafka/clients/consumer/internals/OffsetsForLeaderEpochClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,7 +80,7 @@ protected OffsetForEpochResult handleResponse(
topicPartition, epochEndOffset.endOffset(), epochEndOffset.leaderEpoch());
endOffsets.put(topicPartition, epochEndOffset);
break;
case NOT_LEADER_FOR_PARTITION:
case NOT_LEADER_OR_FOLLOWER:
case REPLICA_NOT_AVAILABLE:
case KAFKA_STORAGE_ERROR:
case OFFSET_NOT_AVAILABLE:
Expand Down
4 changes: 3 additions & 1 deletion clients/src/main/java/org/apache/kafka/common/errors/NotLeaderForPartitionException.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,8 +17,10 @@
package org.apache.kafka.common.errors;

/**
* This server is not the leader for the given partition
* This server is not the leader for the given partition.
* @deprecated since 2.7. Use {@link NotLeaderOrFollowerException}.
*/
@Deprecated
public class NotLeaderForPartitionException extends InvalidMetadataException {

private static final long serialVersionUID = 1L;
Expand Down
44 changes: 44 additions & 0 deletions clients/src/main/java/org/apache/kafka/common/errors/NotLeaderOrFollowerException.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,44 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.kafka.common.errors;

/**
* This server is not the leader or follower for the given partition.
* This could a transient exception during reassignments.
*/
@SuppressWarnings("deprecation")
public class NotLeaderOrFollowerException extends NotLeaderForPartitionException {

@hachikuji hachikuji Jul 15, 2020

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.

One nitpick on the naming. It seems a little strange that followers might return NOT_LEADER_OR_FOLLOWER for requests which are intended for the leader. I think my suggestion to frame the name around the broker not being the intended target of the request was a little more accurate, though I admit the name was clumsier. Perhaps this class would be a good place to document the semantics? Maybe worth mentioning the Produce/Fetch behavior explicitly.

@ijuma ijuma Jul 16, 2020

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.

Good point.

@rajinisivaram rajinisivaram Jul 16, 2020

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.

Updated javadoc.


private static final long serialVersionUID = 1L;

public NotLeaderOrFollowerException() {
super();
}

public NotLeaderOrFollowerException(String message) {
super(message);
}

public NotLeaderOrFollowerException(Throwable cause) {
super(cause);
}

public NotLeaderOrFollowerException(String message, Throwable cause) {
super(message, cause);
}

}
6 changes: 3 additions & 3 deletions clients/src/main/java/org/apache/kafka/common/protocol/Errors.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,7 +72,7 @@
import org.apache.kafka.common.errors.NotCoordinatorException;
import org.apache.kafka.common.errors.NotEnoughReplicasAfterAppendException;
import org.apache.kafka.common.errors.NotEnoughReplicasException;
import org.apache.kafka.common.errors.NotLeaderForPartitionException;
import org.apache.kafka.common.errors.NotLeaderOrFollowerException;
import org.apache.kafka.common.errors.OffsetMetadataTooLarge;
import org.apache.kafka.common.errors.OffsetNotAvailableException;
import org.apache.kafka.common.errors.OffsetOutOfRangeException;
Expand Down Expand Up @@ -139,8 +139,8 @@ public enum Errors {
InvalidFetchSizeException::new),
LEADER_NOT_AVAILABLE(5, "There is no leader for this topic-partition as we are in the middle of a leadership election.",
LeaderNotAvailableException::new),
NOT_LEADER_FOR_PARTITION(6, "This server is not the leader for that topic-partition.",
NotLeaderForPartitionException::new),
NOT_LEADER_OR_FOLLOWER(6, "This server is not the leader or follower for that topic-partition.",

@hachikuji hachikuji Jul 15, 2020

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.

This would also be a good place to be clearer about the interpretation of this error. Maybe something like this?

For requests intended only for the leader, this error indicates that the broker is not the current leader. For requests intended for any replica, this error indicates that the broker is not a replica of the topic partition.

NotLeaderOrFollowerException::new),
REQUEST_TIMED_OUT(7, "The request timed out.",
TimeoutException::new),
BROKER_NOT_AVAILABLE(8, "The broker is not available.",
Comment thread
hachikuji marked this conversation as resolved.
Expand Down
3 changes: 2 additions & 1 deletion clients/src/main/java/org/apache/kafka/common/requests/AlterReplicaLogDirsResponse.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,8 @@
*
* {@link Errors#LOG_DIR_NOT_FOUND}
* {@link Errors#KAFKA_STORAGE_ERROR}
* {@link Errors#REPLICA_NOT_AVAILABLE}
* {@link Errors#REPLICA_NOT_AVAILABLE} from brokers with inter-broker protocol version < 2.7
* {@link Errors#NOT_LEADER_OR_FOLLOWER} from brokers with inter-broker protocol version >= 2.7
* {@link Errors#UNKNOWN_SERVER_ERROR}
*/
public class AlterReplicaLogDirsResponse extends AbstractResponse {
Expand Down
2 changes: 1 addition & 1 deletion clients/src/main/java/org/apache/kafka/common/requests/DeleteRecordsResponse.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,7 @@ public class DeleteRecordsResponse extends AbstractResponse {
*
* OFFSET_OUT_OF_RANGE (1)
* UNKNOWN_TOPIC_OR_PARTITION (3)
* NOT_LEADER_FOR_PARTITION (6)
* NOT_LEADER_OR_FOLLOWER (6)
* REQUEST_TIMED_OUT (7)
* UNKNOWN (-1)
*/
Expand Down
2 changes: 1 addition & 1 deletion clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -146,7 +146,7 @@ public class FetchRequest extends AbstractRequest {
TOPICS_V5);

// V6 bumped up to indicate that the client supports KafkaStorageException. The KafkaStorageException will be
// translated to NotLeaderForPartitionException in the response if version <= 5
// translated to NotLeaderOrFollowerException in the response if version <= 5
private static final Schema FETCH_REQUEST_V6 = FETCH_REQUEST_V5;

// V7 added incremental fetch requests.
Expand Down
10 changes: 5 additions & 5 deletions clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,8 +58,8 @@
*
* - {@link Errors#OFFSET_OUT_OF_RANGE} If the fetch offset is out of range for a requested partition
* - {@link Errors#TOPIC_AUTHORIZATION_FAILED} If the user does not have READ access to a requested topic
* - {@link Errors#REPLICA_NOT_AVAILABLE} If the request is received by a broker which is not a replica
* - {@link Errors#NOT_LEADER_FOR_PARTITION} If the broker is not a leader and either the provided leader epoch
* - {@link Errors#REPLICA_NOT_AVAILABLE} If the request is received by a broker with version 2.4 to 2.6 which is not a replica

@hachikuji hachikuji Jul 16, 2020

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.

Is the note about the range from 2.4 to 2.6 correct? I think this error has always been possible in the case of a reassignment.

@rajinisivaram rajinisivaram Jul 17, 2020

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.

Updated to say < 2.6.

* - {@link Errors#NOT_LEADER_OR_FOLLOWER} If the broker is not a leader or follower and either the provided leader epoch
* matches the known leader epoch on the broker or is empty
* - {@link Errors#FENCED_LEADER_EPOCH} If the epoch is lower than the broker's epoch
* - {@link Errors#UNKNOWN_LEADER_EPOCH} If the epoch is larger than the broker's epoch
Expand Down Expand Up @@ -190,7 +190,7 @@ public class FetchResponse<T extends BaseRecords> extends AbstractResponse {
new Field(RESPONSES_KEY_NAME, new ArrayOf(FETCH_RESPONSE_TOPIC_V5)));

// V6 bumped up to indicate that the client supports KafkaStorageException. The KafkaStorageException will
// be translated to NotLeaderForPartitionException in the response if version <= 5
// be translated to NotLeaderOrFollowerException in the response if version <= 5
private static final Schema FETCH_RESPONSE_V6 = FETCH_RESPONSE_V5;

// V7 added incremental fetch responses and a top-level error code.
Expand Down Expand Up @@ -549,9 +549,9 @@ private static <T extends BaseRecords> Struct toStruct(short version, int thrott
// If consumer sends FetchRequest V5 or earlier, the client library is not guaranteed to recognize the error code
// for KafkaStorageException. In this case the client library will translate KafkaStorageException to
// UnknownServerException which is not retriable. We can ensure that consumer will update metadata and retry
// by converting the KafkaStorageException to NotLeaderForPartitionException in the response if FetchRequest version <= 5
// by converting the KafkaStorageException to NotLeaderOrFollowerException in the response if FetchRequest version <= 5
if (errorCode == Errors.KAFKA_STORAGE_ERROR.code() && version <= 5)
errorCode = Errors.NOT_LEADER_FOR_PARTITION.code();
errorCode = Errors.NOT_LEADER_OR_FOLLOWER.code();
Struct partitionData = topicData.instance(PARTITIONS_KEY_NAME);
Struct partitionDataHeader = partitionData.instance(PARTITION_HEADER_KEY_NAME);
partitionDataHeader.set(PARTITION_ID, partitionEntry.getKey());
Expand Down
4 changes: 2 additions & 2 deletions clients/src/main/java/org/apache/kafka/common/requests/ListOffsetResponse.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,8 +44,8 @@
*
* - {@link Errors#UNSUPPORTED_FOR_MESSAGE_FORMAT} If the message format does not support lookup by timestamp
* - {@link Errors#TOPIC_AUTHORIZATION_FAILED} If the user does not have DESCRIBE access to a requested topic
* - {@link Errors#REPLICA_NOT_AVAILABLE} If the request is received by a broker which is not a replica
* - {@link Errors#NOT_LEADER_FOR_PARTITION} If the broker is not a leader and either the provided leader epoch
* - {@link Errors#REPLICA_NOT_AVAILABLE} If the request is received by a broker with version 2.4 to 2.6 which is not a replica
* - {@link Errors#NOT_LEADER_OR_FOLLOWER} If the broker is not a leader or follower and either the provided leader epoch
* matches the known leader epoch on the broker or is empty
* - {@link Errors#FENCED_LEADER_EPOCH} If the epoch is lower than the broker's epoch
* - {@link Errors#UNKNOWN_LEADER_EPOCH} If the epoch is larger than the broker's epoch
Expand Down
4 changes: 2 additions & 2 deletions clients/src/main/java/org/apache/kafka/common/requests/OffsetsForLeaderEpochResponse.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,8 +41,8 @@
* Possible error codes:
*
* - {@link Errors#TOPIC_AUTHORIZATION_FAILED} If the user does not have DESCRIBE access to a requested topic
* - {@link Errors#REPLICA_NOT_AVAILABLE} If the request is received by a broker which is not a replica
* - {@link Errors#NOT_LEADER_FOR_PARTITION} If the broker is not a leader and either the provided leader epoch
* - {@link Errors#REPLICA_NOT_AVAILABLE} If the request is received by a broker with version 2.4 to 2.6 which is not a replica
* - {@link Errors#NOT_LEADER_OR_FOLLOWER} If the broker is not a leader or follower and either the provided leader epoch
* matches the known leader epoch on the broker or is empty
* - {@link Errors#FENCED_LEADER_EPOCH} If the epoch is lower than the broker's epoch
* - {@link Errors#UNKNOWN_LEADER_EPOCH} If the epoch is larger than the broker's epoch
Expand Down
2 changes: 1 addition & 1 deletion clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,7 +99,7 @@ public class ProduceRequest extends AbstractRequest {
/**
* The body of PRODUCE_REQUEST_V4 is the same as PRODUCE_REQUEST_V3.
* The version number is bumped up to indicate that the client supports KafkaStorageException.
* The KafkaStorageException will be translated to NotLeaderForPartitionException in the response if version <= 3
* The KafkaStorageException will be translated to NotLeaderOrFollowerException in the response if version <= 3
*/
private static final Schema PRODUCE_REQUEST_V4 = PRODUCE_REQUEST_V3;

Expand Down
8 changes: 4 additions & 4 deletions clients/src/main/java/org/apache/kafka/common/requests/ProduceResponse.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,7 +56,7 @@ public class ProduceResponse extends AbstractResponse {
*
* {@link Errors#CORRUPT_MESSAGE}
* {@link Errors#UNKNOWN_TOPIC_OR_PARTITION}
* {@link Errors#NOT_LEADER_FOR_PARTITION}
* {@link Errors#NOT_LEADER_OR_FOLLOWER}
* {@link Errors#MESSAGE_TOO_LARGE}
* {@link Errors#INVALID_TOPIC_EXCEPTION}
* {@link Errors#RECORD_LIST_TOO_LARGE}
Expand Down Expand Up @@ -126,7 +126,7 @@ public class ProduceResponse extends AbstractResponse {
/**
* The body of PRODUCE_RESPONSE_V4 is the same as PRODUCE_RESPONSE_V3.
* The version number is bumped up to indicate that the client supports KafkaStorageException.
* The KafkaStorageException will be translated to NotLeaderForPartitionException in the response if version <= 3
* The KafkaStorageException will be translated to NotLeaderOrFollowerException in the response if version <= 3
*/
private static final Schema PRODUCE_RESPONSE_V4 = PRODUCE_RESPONSE_V3;

Expand Down Expand Up @@ -265,9 +265,9 @@ protected Struct toStruct(short version) {
// If producer sends ProduceRequest V3 or earlier, the client library is not guaranteed to recognize the error code
// for KafkaStorageException. In this case the client library will translate KafkaStorageException to
// UnknownServerException which is not retriable. We can ensure that producer will update metadata and retry
// by converting the KafkaStorageException to NotLeaderForPartitionException in the response if ProduceRequest version <= 3
// by converting the KafkaStorageException to NotLeaderOrFollowerException in the response if ProduceRequest version <= 3
if (errorCode == Errors.KAFKA_STORAGE_ERROR.code() && version <= 3)
errorCode = Errors.NOT_LEADER_FOR_PARTITION.code();
errorCode = Errors.NOT_LEADER_OR_FOLLOWER.code();
Struct partStruct = topicData.instance(PARTITION_RESPONSES_KEY_NAME)
.set(PARTITION_ID, partitionEntry.getKey())
.set(ERROR_CODE, errorCode)
Expand Down
2 changes: 1 addition & 1 deletion clients/src/main/java/org/apache/kafka/common/requests/WriteTxnMarkersResponse.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,7 +37,7 @@
* - {@link Errors#CORRUPT_MESSAGE}
* - {@link Errors#INVALID_PRODUCER_EPOCH}
* - {@link Errors#UNKNOWN_TOPIC_OR_PARTITION}
* - {@link Errors#NOT_LEADER_FOR_PARTITION}
* - {@link Errors#NOT_LEADER_OR_FOLLOWER}
* - {@link Errors#MESSAGE_TOO_LARGE}
* - {@link Errors#RECORD_LIST_TOO_LARGE}
* - {@link Errors#NOT_ENOUGH_REPLICAS}
Expand Down
12 changes: 6 additions & 6 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 @@ -54,7 +54,7 @@
import org.apache.kafka.common.errors.LeaderNotAvailableException;
import org.apache.kafka.common.errors.LogDirNotFoundException;
import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kafka.common.errors.NotLeaderForPartitionException;
import org.apache.kafka.common.errors.NotLeaderOrFollowerException;
import org.apache.kafka.common.errors.OffsetOutOfRangeException;
import org.apache.kafka.common.errors.SaslAuthenticationException;
import org.apache.kafka.common.errors.SecurityDisabledException;
Expand Down Expand Up @@ -1201,7 +1201,7 @@ public void testDeleteRecords() throws Exception {
new DeleteRecordsResponseData.DeleteRecordsPartitionResult()
.setPartitionIndex(myTopicPartition3.partition())
.setLowWatermark(DeleteRecordsResponse.INVALID_LOW_WATERMARK)
.setErrorCode(Errors.NOT_LEADER_FOR_PARTITION.code()),
.setErrorCode(Errors.NOT_LEADER_OR_FOLLOWER.code()),
new DeleteRecordsResponseData.DeleteRecordsPartitionResult()
.setPartitionIndex(myTopicPartition4.partition())
.setLowWatermark(DeleteRecordsResponse.INVALID_LOW_WATERMARK)
Expand Down Expand Up @@ -1270,7 +1270,7 @@ public void testDeleteRecords() throws Exception {
myTopicPartition3Result.get();
fail("get() should throw ExecutionException");
} catch (ExecutionException e1) {
assertTrue(e1.getCause() instanceof NotLeaderForPartitionException);
assertTrue(e1.getCause() instanceof NotLeaderOrFollowerException);
}

// "unknown topic or partition" failure on records deletion for partition 4
Expand Down Expand Up @@ -3270,7 +3270,7 @@ public void testListOffsetsWithMultiplePartitionsLeaderChange() throws Exception
env.kafkaClient().prepareResponse(prepareMetadataResponse(oldCluster, Errors.NONE));

Map<TopicPartition, PartitionData> responseData = new HashMap<>();
responseData.put(tp0, new PartitionData(Errors.NOT_LEADER_FOR_PARTITION, -1L, 345L, Optional.of(543)));
responseData.put(tp0, new PartitionData(Errors.NOT_LEADER_OR_FOLLOWER, -1L, 345L, Optional.of(543)));
responseData.put(tp1, new PartitionData(Errors.LEADER_NOT_AVAILABLE, -2L, 123L, Optional.of(456)));
env.kafkaClient().prepareResponseFrom(new ListOffsetResponse(responseData), node0);

Expand Down Expand Up @@ -3328,10 +3328,10 @@ public void testListOffsetsWithLeaderChange() throws Exception {
env.kafkaClient().prepareResponse(prepareMetadataResponse(oldCluster, Errors.NONE));

Map<TopicPartition, PartitionData> responseData = new HashMap<>();
responseData.put(tp0, new PartitionData(Errors.NOT_LEADER_FOR_PARTITION, -1L, 345L, Optional.of(543)));
responseData.put(tp0, new PartitionData(Errors.NOT_LEADER_OR_FOLLOWER, -1L, 345L, Optional.of(543)));
env.kafkaClient().prepareResponseFrom(new ListOffsetResponse(responseData), node0);

// updating leader from node0 to node1 and metadata refresh because of NOT_LEADER_FOR_PARTITION
// updating leader from node0 to node1 and metadata refresh because of NOT_LEADER_OR_FOLLOWER
final PartitionInfo newPartitionInfo = new PartitionInfo("foo", 0, node1,
new Node[]{node0, node1, node2}, new Node[]{node0, node1, node2});
final Cluster newCluster = new Cluster("mockClusterId", nodes, singletonList(newPartitionInfo),
Expand Down
Loading