Skip to content

KAFKA-13888: Addition of Information in DescribeQuorumResponse #12508

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
hachikuji merged 17 commits into from
Aug 19, 2022
Merged

KAFKA-13888: Addition of Information in DescribeQuorumResponse #12508

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
17 commits
Select commit Hold shift + click to select a range
c5b4323
KAFKA-13888: Addition of Information in DescribeQuorumResponse
niket-goel Aug 12, 2022
19af2c2
Merge branch 'trunk' into kafka-13888-describe-quorum-fields
niket-goel Aug 16, 2022
ddbfbea
Addressing Review Comments
niket-goel Aug 16, 2022
0d70f91
More Refactoring based on review comments
niket-goel Aug 16, 2022
73e034e
Styling Changes and some test improvement based on review
niket-goel Aug 16, 2022
5de54cd
Removing unused variable from test method
niket-goel Aug 16, 2022
9322538
Removing an unused argument from updateEndOffset
niket-goel Aug 16, 2022
36f4f75
more refactoring
niket-goel Aug 17, 2022
8630b3f
Merge branch 'trunk' into kafka-13888-describe-quorum-fields
niket-goel Aug 17, 2022
b1c2b1e
Removing check for Kafka-13940 in test as it has been fixed
niket-goel Aug 17, 2022
9a55624
Fixing build failure and removing unnecessary catch
niket-goel Aug 17, 2022
8a105d2
Fixing failure in DescribeQuorumRequestTest
niket-goel Aug 18, 2022
0d79978
Merge branch 'trunk' into kafka-13888-describe-quorum-fields
niket-goel Aug 18, 2022
b00e66c
Fixing breaking test in KafkaRaftClientTest
niket-goel Aug 18, 2022
a12013c
Removing falky integration test
niket-goel Aug 19, 2022
02dc7ad
Revert "Removing falky integration test"
niket-goel Aug 19, 2022
d1936ab
Return NOT_CONTROLLER if forwarded DescribeQuorum has NOT_LEADER error
niket-goel Aug 19, 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
63 changes: 61 additions & 2 deletions core/src/test/scala/integration/kafka/server/KRaftClusterTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@ import kafka.network.SocketServer
import kafka.server.IntegrationTestUtils.connectAndReceive
import kafka.testkit.{BrokerNode, KafkaClusterTestKit, TestKitNodes}
import kafka.utils.TestUtils
import org.apache.kafka.clients.admin.{Admin, AlterConfigOp, Config, ConfigEntry, NewPartitionReassignment, NewTopic}
import org.apache.kafka.clients.admin.{Admin, AdminClientConfig, AlterConfigOp, Config, ConfigEntry, DescribeMetadataQuorumOptions, NewPartitionReassignment, NewTopic}
import org.apache.kafka.common.{TopicPartition, TopicPartitionInfo}
import org.apache.kafka.common.message.DescribeClusterRequestData
import org.apache.kafka.common.network.ListenerName
Expand All @@ -32,10 +32,11 @@ import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.{Tag, Test, Timeout}

import java.util
import java.util.{Arrays, Collections, Optional}
import java.util.{Arrays, Collections, Optional, OptionalLong, Properties}
import org.apache.kafka.clients.admin.AlterConfigOp.OpType
import org.apache.kafka.common.config.ConfigResource
import org.apache.kafka.common.config.ConfigResource.Type
import org.apache.kafka.common.errors.InvalidRequestException
import org.apache.kafka.common.protocol.Errors._
import org.apache.kafka.image.ClusterImage
import org.apache.kafka.server.common.MetadataVersion
Expand Down Expand Up @@ -778,4 +779,62 @@ class KRaftClusterTest {
cluster.close()
}
}
def createAdminClient(cluster: KafkaClusterTestKit, useController: Boolean): Admin = {
var props: Properties = null
props = cluster.clientProperties()
props.put(AdminClientConfig.CLIENT_ID_CONFIG, this.getClass.getName)
Admin.create(props)
}

@Test
def testDescribeQuorumRequestToBrokers() : Unit = {
val cluster = new KafkaClusterTestKit.Builder(
new TestKitNodes.Builder().
setNumBrokerNodes(4).
setNumControllerNodes(3).build()).build()
try {
cluster.format
cluster.startup
for (i <- 0 to 3) {
TestUtils.waitUntilTrue(() => cluster.brokers.get(i).brokerState == BrokerState.RUNNING,
"Broker Never started up")
}
val admin = createAdminClient(cluster, false)
try {
val quorumState = admin.describeMetadataQuorum(new DescribeMetadataQuorumOptions)
val quorumInfo = quorumState.quorumInfo.get()

assertEquals(4, quorumInfo.observers.size)
assertEquals(3, quorumInfo.voters.size)

@hachikuji hachikuji Aug 15, 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.

A more straightforward way to assert the voter/observer sets is in DescribeQuorumRequestTest:

      val voterData = partitionData.currentVoters.asScala
      assertEquals(cluster.controllerIds().asScala, voterData.map(_.replicaId).toSet);

      val observerData = partitionData.observers.asScala
      assertEquals(cluster.brokerIds().asScala, observerData.map(_.replicaId).toSet);

@niket-goel niket-goel Aug 16, 2022

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.

That check already exists in the DescribeQuorumRequestTest. The count validation here is just a sanity check.

@hachikuji hachikuji Aug 16, 2022 •
edited
Loading

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.

We are checking the size here and all of the individual replicaIds below. For example:

assertTrue(-1 < observer.replicaId && 4 > observer.replicaId,
            s"Observer ID ${observer.replicaId} was not within expected range.")

I am just offering a more concise and complete way to do that.

@hachikuji hachikuji Aug 16, 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.

You may have missed my comment above. The nice thing about asserting the set directly is that it ensures that all expected voters/observers are present in the result.

assertTrue(2999 < quorumInfo.leaderId && 3003 > quorumInfo.leaderId,

@hachikuji hachikuji Aug 16, 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.

We could make this assertion a little more generic by asserting that quorumInfo.leaderId is contained in cluster.controllers.asScala.keySet.

s"Leader ID ${quorumInfo.leaderId} was not within expected range.")

quorumInfo.observers.forEach { observer =>
assertTrue(-1 < observer.replicaId && 4 > observer.replicaId,
s"Observer ID ${observer.replicaId} was not within expected range.")
assertTrue(0 < observer.logEndOffset,
s"logEndOffset for observer with ID ${observer.replicaId} was ${observer.logEndOffset}")
assertNotEquals(OptionalLong.empty(), observer.lastFetchTimeMs)
assertNotEquals(OptionalLong.empty(), observer.lastCaughtUpTimeMs)
}

quorumInfo.voters.forEach { voter =>
assertTrue(2999 < voter.replicaId && 3003 > voter.replicaId,
s"Voter ID ${voter.replicaId} was not within expected range.")
assertTrue(0 < voter.logEndOffset,
s"logEndOffset for voter with ID ${voter.replicaId} was ${voter.logEndOffset}")
assertNotEquals(OptionalLong.empty(), voter.lastFetchTimeMs)
assertNotEquals(OptionalLong.empty(), voter.lastCaughtUpTimeMs)
}
} catch {
case _: InvalidRequestException => log.error("Hit Kafka-13490. Claim test succeeded")
case t: Throwable => throw t

@hachikuji hachikuji Aug 17, 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: if we're just re-throwing, we don't need to catch

} finally {
admin.close()
}
} finally {
cluster.close()
}
}

}
16 changes: 3 additions & 13 deletions raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,6 @@
import org.apache.kafka.common.message.BeginQuorumEpochResponseData;
import org.apache.kafka.common.message.DescribeQuorumRequestData;
import org.apache.kafka.common.message.DescribeQuorumResponseData;
import org.apache.kafka.common.message.DescribeQuorumResponseData.ReplicaState;
import org.apache.kafka.common.message.EndQuorumEpochRequestData;
import org.apache.kafka.common.message.EndQuorumEpochResponseData;
import org.apache.kafka.common.message.FetchRequestData;
Expand Down Expand Up @@ -95,7 +94,6 @@
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Supplier;
import java.util.stream.Collectors;

import static java.util.concurrent.CompletableFuture.completedFuture;
import static org.apache.kafka.raft.RaftUtil.hasValidTopicPartition;
Expand Down Expand Up @@ -1016,7 +1014,7 @@ private FetchResponseData tryCompleteFetchRequest(
if (validOffsetAndEpoch.kind() == ValidOffsetAndEpoch.Kind.VALID) {
LogFetchInfo info = log.read(fetchOffset, Isolation.UNCOMMITTED);

if (state.updateReplicaState(replicaId, currentTimeMs, info.startOffsetMetadata)) {
if (state.updateReplicaState(replicaId, currentTimeMs, info.startOffsetMetadata, log.endOffset().offset)) {
onUpdateLeaderHighWatermark(state, currentTimeMs);
}

Expand Down Expand Up @@ -1179,8 +1177,8 @@ private DescribeQuorumResponseData handleDescribeQuorumRequest(
leaderState.localId(),
leaderState.epoch(),
leaderState.highWatermark().isPresent() ? leaderState.highWatermark().get().offset : -1,
convertToReplicaStates(leaderState.getVoterEndOffsets()),
convertToReplicaStates(leaderState.getObserverStates(currentTimeMs))
leaderState.quorumResponseVoterStates(currentTimeMs),
leaderState.quorumResponseObserverStates(currentTimeMs)
);
}

Expand Down Expand Up @@ -1418,14 +1416,6 @@ private boolean handleFetchSnapshotResponse(
return true;
}

List<ReplicaState> convertToReplicaStates(Map<Integer, Long> replicaEndOffsets) {
return replicaEndOffsets.entrySet().stream()
.map(entry -> new ReplicaState()
.setReplicaId(entry.getKey())
.setLogEndOffset(entry.getValue()))
.collect(Collectors.toList());
}

private boolean hasConsistentLeader(int epoch, OptionalInt leaderId) {
// Only elected leaders are sent in the request/response header, so if we have an elected
// leaderId, it should be consistent with what is in the message.
Expand Down
113 changes: 95 additions & 18 deletions raft/src/main/java/org/apache/kafka/raft/LeaderState.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
*/
package org.apache.kafka.raft;

import org.apache.kafka.common.message.DescribeQuorumResponseData;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.raft.internals.BatchAccumulator;
import org.slf4j.Logger;
Expand All @@ -31,6 +32,7 @@
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.OptionalInt;
import java.util.OptionalLong;
import java.util.Set;
import java.util.stream.Collectors;
Expand Down Expand Up @@ -205,10 +207,10 @@ private boolean updateHighWatermark() {
/**
* Update the local replica state.
*
* See {@link #updateReplicaState(int, long, LogOffsetMetadata)}
* See {@link #updateReplicaState(int, long, LogOffsetMetadata, Long)}
*/
public boolean updateLocalState(long fetchTimestamp, LogOffsetMetadata logOffsetMetadata) {
return updateReplicaState(localId, fetchTimestamp, logOffsetMetadata);
return updateReplicaState(localId, fetchTimestamp, logOffsetMetadata, logOffsetMetadata.offset);
}

/**
Expand All @@ -217,20 +219,36 @@ public boolean updateLocalState(long fetchTimestamp, LogOffsetMetadata logOffset
* @param replicaId replica id
* @param fetchTimestamp fetch timestamp
* @param logOffsetMetadata new log offset and metadata
* @param leaderLogEndOffset current log end offset of the leader
* @return true if the high watermark is updated too
*/
public boolean updateReplicaState(int replicaId,
long fetchTimestamp,
LogOffsetMetadata logOffsetMetadata) {
public boolean updateReplicaState(
int replicaId,
long fetchTimestamp,
LogOffsetMetadata logOffsetMetadata,
Long leaderLogEndOffset
Comment thread
niket-goel marked this conversation as resolved.
Outdated
) {
// Ignore fetches from negative replica id, as it indicates
// the fetch is from non-replica. For example, a consumer.
if (replicaId < 0) {
return false;
}

ReplicaState state = getReplicaState(replicaId);
state.updateFetchTimestamp(fetchTimestamp);
return updateEndOffset(state, logOffsetMetadata);

// Only proceed with updating the states if the offset update is valid
verifyEndOffsetUpdate(state, logOffsetMetadata);

// Update the Last CaughtUp Time
if (logOffsetMetadata.offset >= leaderLogEndOffset) {
state.updateLastCaughtUpTimestamp(fetchTimestamp);

@hachikuji hachikuji Aug 15, 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.

We do these updates before the call to updateEndOffset, which may raise an exception (same thing for updateFetchTimestamp which is an existing issue). I think it would be better to update state only after we have confirmed the update is valid.

} else if (logOffsetMetadata.offset >= state.lastFetchLeaderLogEndOffset.orElse(-1L)) {
state.updateLastCaughtUpTimestamp(state.lastFetchTimestamp.orElse(-1L));
}

state.updateFetchTimestamp(fetchTimestamp, leaderLogEndOffset);
return updateEndOffset(state, logOffsetMetadata, false);

Comment thread
niket-goel marked this conversation as resolved.
Outdated
}

public List<Integer> nonLeaderVotersByDescendingFetchOffset() {
Expand All @@ -246,20 +264,30 @@ private List<ReplicaState> followersByDescendingFetchOffset() {
.collect(Collectors.toList());
}

private boolean updateEndOffset(ReplicaState state,
LogOffsetMetadata endOffsetMetadata) {
private void verifyEndOffsetUpdate(
ReplicaState state,
LogOffsetMetadata endOffsetMetadata
) {
state.endOffset.ifPresent(currentEndOffset -> {
if (currentEndOffset.offset > endOffsetMetadata.offset) {
if (state.nodeId == localId) {
throw new IllegalStateException("Detected non-monotonic update of local " +
"end offset: " + currentEndOffset.offset + " -> " + endOffsetMetadata.offset);
"end offset: " + currentEndOffset.offset + " -> " + endOffsetMetadata.offset);
Comment thread
niket-goel marked this conversation as resolved.
Outdated
} else {
log.warn("Detected non-monotonic update of fetch offset from nodeId {}: {} -> {}",
state.nodeId, currentEndOffset.offset, endOffsetMetadata.offset);
state.nodeId, currentEndOffset.offset, endOffsetMetadata.offset);
}
}
});

}
private boolean updateEndOffset(
ReplicaState state,
Comment thread
niket-goel marked this conversation as resolved.
Outdated
LogOffsetMetadata endOffsetMetadata,
boolean verifyUpdate
) {
if (verifyUpdate) {
verifyEndOffsetUpdate(state, endOffsetMetadata);
}
state.endOffset = Optional.of(endOffsetMetadata);
state.hasAcknowledgedLeader = true;
return isVoter(state.nodeId) && updateHighWatermark();
Expand Down Expand Up @@ -290,13 +318,30 @@ private ReplicaState getReplicaState(int remoteNodeId) {
return state;
}

Map<Integer, Long> getVoterEndOffsets() {
return getReplicaEndOffsets(voterStates);
List<DescribeQuorumResponseData.ReplicaState> quorumResponseVoterStates(long currentTimeMs) {
return quorumResponseReplicaStates(voterStates, OptionalInt.of(localId), currentTimeMs);
}

Map<Integer, Long> getObserverStates(final long currentTimeMs) {
List<DescribeQuorumResponseData.ReplicaState> quorumResponseObserverStates(long currentTimeMs) {
clearInactiveObservers(currentTimeMs);
return getReplicaEndOffsets(observerStates);
return quorumResponseReplicaStates(observerStates, OptionalInt.empty(), currentTimeMs);
}

private static <R extends ReplicaState> List<DescribeQuorumResponseData.ReplicaState> quorumResponseReplicaStates(
Comment thread
niket-goel marked this conversation as resolved.
Outdated
Map<Integer, R> state,
OptionalInt leaderId,
Comment thread
niket-goel marked this conversation as resolved.
Outdated
long currentTimeMs) {
Map<Integer, Long> replicaEndOffsets = getReplicaEndOffsets(state);
Comment thread
hachikuji marked this conversation as resolved.
Outdated
Map<Integer, Long> replicaLastFetchTimes = getReplicaLastFetchTimes(state);
Map<Integer, Long> replicaLastCaughtUpTimes = getReplicaLastCaughtUpTimes(state, leaderId, currentTimeMs);

return replicaEndOffsets.entrySet().stream()
.map(entry -> new DescribeQuorumResponseData.ReplicaState()
.setReplicaId(entry.getKey())
.setLogEndOffset(entry.getValue())
.setLastFetchTimestamp(replicaLastFetchTimes.get(entry.getKey()))
.setLastCaughtUpTimestamp(replicaLastCaughtUpTimes.get(entry.getKey())))
.collect(Collectors.toList());
}

private static <R extends ReplicaState> Map<Integer, Long> getReplicaEndOffsets(
Expand All @@ -308,6 +353,25 @@ private static <R extends ReplicaState> Map<Integer, Long> getReplicaEndOffsets(
);
}

private static <R extends ReplicaState> Map<Integer, Long> getReplicaLastFetchTimes(
Map<Integer, R> replicaStates) {
return replicaStates.entrySet().stream()
.collect(Collectors.toMap(Map.Entry::getKey,
e -> e.getValue().lastFetchTimestamp.orElse(-1L))
);
}

private static <R extends ReplicaState> Map<Integer, Long> getReplicaLastCaughtUpTimes(
Map<Integer, R> replicaStates,
OptionalInt leaderId,
final long currentTimeMs
) {
return replicaStates.entrySet().stream()
.collect(Collectors.toMap(Map.Entry::getKey,
e -> e.getKey() == leaderId.orElse(-1) ? currentTimeMs : e.getValue().lastCaughtUpTimestamp.orElse(-1L))
);
}

private void clearInactiveObservers(final long currentTimeMs) {
observerStates.entrySet().removeIf(
integerReplicaStateEntry ->
Expand All @@ -323,19 +387,30 @@ private static class ReplicaState implements Comparable<ReplicaState> {
final int nodeId;
Optional<LogOffsetMetadata> endOffset;
OptionalLong lastFetchTimestamp;
OptionalLong lastFetchLeaderLogEndOffset;
OptionalLong lastCaughtUpTimestamp;
boolean hasAcknowledgedLeader;

public ReplicaState(int nodeId, boolean hasAcknowledgedLeader) {
this.nodeId = nodeId;
this.endOffset = Optional.empty();
this.lastFetchTimestamp = OptionalLong.empty();
this.lastFetchLeaderLogEndOffset = OptionalLong.empty();
this.lastCaughtUpTimestamp = OptionalLong.empty();
this.hasAcknowledgedLeader = hasAcknowledgedLeader;
}

void updateFetchTimestamp(long currentFetchTimeMs) {
void updateFetchTimestamp(long currentFetchTimeMs, long leaderLogEndOffset) {
// To be resilient to system time shifts we do not strictly
// require the timestamp be monotonically increasing.
lastFetchTimestamp = OptionalLong.of(Math.max(lastFetchTimestamp.orElse(-1L), currentFetchTimeMs));
lastFetchLeaderLogEndOffset = OptionalLong.of(leaderLogEndOffset);
}

void updateLastCaughtUpTimestamp(long lastCaughtUpTime) {
// This value relies on the fetch timestamp which does not
// require monotonicity
lastCaughtUpTimestamp = OptionalLong.of(Math.max(lastCaughtUpTimestamp.orElse(-1L), lastCaughtUpTime));
}

@Override
Expand All @@ -353,10 +428,12 @@ else if (!that.endOffset.isPresent())
@Override
public String toString() {
return String.format(
"ReplicaState(nodeId=%d, endOffset=%s, lastFetchTimestamp=%s, hasAcknowledgedLeader=%s)",
"ReplicaState(nodeId=%d, endOffset=%s, lastFetchTimestamp=%s, " +
" lastCaughtUpTimestamp=%s, hasAcknowledgedLeader=%s)",
Comment thread
niket-goel marked this conversation as resolved.
Outdated
nodeId,
endOffset,
lastFetchTimestamp,
lastCaughtUpTimestamp,
hasAcknowledgedLeader
);
}
Expand Down
Loading