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 1 commit
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
22 changes: 9 additions & 13 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 @@ -804,28 +804,24 @@ class KRaftClusterTest {
val quorumState = admin.describeMetadataQuorum(new DescribeMetadataQuorumOptions)
val quorumInfo = quorumState.quorumInfo.get()

assertEquals(4, quorumInfo.observers.size)
assertEquals(3, quorumInfo.voters.size)
assertEquals(cluster.controllers.asScala.keySet, quorumInfo.voters.asScala.map(_.replicaId).toSet)
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)
}

assertEquals(cluster.brokers.asScala.keySet, quorumInfo.observers.asScala.map(_.replicaId).toSet)
quorumInfo.observers.forEach { observer =>
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)
}
} 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

Expand Down
38 changes: 19 additions & 19 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 @@ -33,7 +33,6 @@
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 @@ -208,7 +207,7 @@ private boolean updateHighWatermark() {
/**
* Update the local replica state.
*
* See {@link #updateReplicaState(int, long, LogOffsetMetadata, Long)}
* See {@link #updateReplicaState(int, long, LogOffsetMetadata, long)}
*/
public boolean updateLocalState(long fetchTimestamp, LogOffsetMetadata logOffsetMetadata) {
return updateReplicaState(localId, fetchTimestamp, logOffsetMetadata, logOffsetMetadata.offset);
Expand All @@ -224,10 +223,10 @@ public boolean updateLocalState(long fetchTimestamp, LogOffsetMetadata logOffset
* @return true if the high watermark is updated too
*/
public boolean updateReplicaState(
int replicaId,
long fetchTimestamp,
LogOffsetMetadata logOffsetMetadata,
Long leaderLogEndOffset
int replicaId,
long fetchTimestamp,
LogOffsetMetadata logOffsetMetadata,
long leaderLogEndOffset
) {
// Ignore fetches from negative replica id, as it indicates
// the fetch is from non-replica. For example, a consumer.
Expand Down Expand Up @@ -266,8 +265,8 @@ private List<ReplicaState> followersByDescendingFetchOffset() {
}

private void verifyEndOffsetUpdate(
ReplicaState state,
LogOffsetMetadata endOffsetMetadata
ReplicaState state,
LogOffsetMetadata endOffsetMetadata
) {
state.endOffset.ifPresent(currentEndOffset -> {
if (currentEndOffset.offset > endOffsetMetadata.offset) {
Expand All @@ -282,9 +281,9 @@ private void verifyEndOffsetUpdate(
});
}
private boolean updateEndOffset(
ReplicaState state,
LogOffsetMetadata endOffsetMetadata,
boolean verifyUpdate
ReplicaState state,
LogOffsetMetadata endOffsetMetadata,
boolean verifyUpdate

@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.

Why do we need this? Seems like false is the only value we ever pass.

@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.

Wanted to retain the verification step in the method in case we want to use it elsewhere in the future.

) {
if (verifyUpdate) {
verifyEndOffsetUpdate(state, endOffsetMetadata);
Expand Down Expand Up @@ -320,22 +319,23 @@ private ReplicaState getReplicaState(int remoteNodeId) {
}

List<DescribeQuorumResponseData.ReplicaState> quorumResponseVoterStates(long currentTimeMs) {
return quorumResponseReplicaStates(voterStates.values(), OptionalInt.of(localId), currentTimeMs);
return quorumResponseReplicaStates(voterStates.values(), localId, currentTimeMs);
}

List<DescribeQuorumResponseData.ReplicaState> quorumResponseObserverStates(long currentTimeMs) {
clearInactiveObservers(currentTimeMs);
return quorumResponseReplicaStates(observerStates.values(), OptionalInt.empty(), currentTimeMs);
return quorumResponseReplicaStates(observerStates.values(), localId, currentTimeMs);
}

private static <R extends ReplicaState> List<DescribeQuorumResponseData.ReplicaState> quorumResponseReplicaStates(
Collection<R> state,
OptionalInt leaderId,
long currentTimeMs) {
private static List<DescribeQuorumResponseData.ReplicaState> quorumResponseReplicaStates(
Collection<ReplicaState> state,
int leaderId,
long currentTimeMs
) {
return state.stream().map(s -> {
final long lastCaughtUpTimestamp;
final long lastFetchTimestamp;
if (s.nodeId == leaderId.orElse(-1)) {
if (s.nodeId == leaderId) {
lastCaughtUpTimestamp = currentTimeMs;
lastFetchTimestamp = currentTimeMs;
} else {
Expand Down Expand Up @@ -407,7 +407,7 @@ else if (!that.endOffset.isPresent())
public String toString() {
return String.format(
"ReplicaState(nodeId=%d, endOffset=%s, lastFetchTimestamp=%s, " +
" lastCaughtUpTimestamp=%s, hasAcknowledgedLeader=%s)",
"lastCaughtUpTimestamp=%s, hasAcknowledgedLeader=%s)",
nodeId,
endOffset,
lastFetchTimestamp,
Expand Down