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

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
62 changes: 20 additions & 42 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 @@ -26,6 +26,7 @@
import org.apache.kafka.common.record.ControlRecordUtils;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
Expand Down Expand Up @@ -319,57 +320,34 @@ private ReplicaState getReplicaState(int remoteNodeId) {
}

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

List<DescribeQuorumResponseData.ReplicaState> quorumResponseObserverStates(long currentTimeMs) {
clearInactiveObservers(currentTimeMs);
return quorumResponseReplicaStates(observerStates, OptionalInt.empty(), currentTimeMs);
return quorumResponseReplicaStates(observerStates.values(), 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,
Collection<R> state,
OptionalInt leaderId,
Comment thread
niket-goel marked this conversation as resolved.
Outdated
long currentTimeMs) {
Map<Integer, Long> replicaEndOffsets = getReplicaEndOffsets(state);
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(
Map<Integer, R> replicaStates) {
return replicaStates.entrySet().stream()
.collect(Collectors.toMap(Map.Entry::getKey,
e -> e.getValue().endOffset.map(
logOffsetMetadata -> logOffsetMetadata.offset).orElse(-1L))
);
}

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))
);
return state.stream().map(s -> {
final long lastCaughtUpTimestamp;
final long lastFetchTimestamp;
if (s.nodeId == leaderId.orElse(-1)) {
lastCaughtUpTimestamp = currentTimeMs;
lastFetchTimestamp = currentTimeMs;
} else {
lastCaughtUpTimestamp = s.lastCaughtUpTimestamp.orElse(-1);
lastFetchTimestamp = s.lastFetchTimestamp.orElse(-1);
}
return new DescribeQuorumResponseData.ReplicaState()
.setReplicaId(s.nodeId)
.setLogEndOffset(s.endOffset.map(md -> md.offset).orElse(-1L))
.setLastCaughtUpTimestamp(lastCaughtUpTimestamp)
.setLastFetchTimestamp(lastFetchTimestamp);
}).collect(Collectors.toList());
}

private void clearInactiveObservers(final long currentTimeMs) {
Expand Down