[GitHub] [kafka] kamalcph commented on a diff in pull request #13561: KAFKA-14888: Added remote log segments retention functionality based on time and size.
kamalcph commented on code in PR #13561:
URL: https://github.com/apache/kafka/pull/13561#discussion_r1295402786
##
core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java:
##
@@ -1003,6 +1015,134 @@ public RemoteLogMetadataManager
createRemoteLogMetadataManager() {
}
}
+private static RemoteLogSegmentMetadata
createRemoteLogSegmentMetadata(long startOffset, long endOffset, Map segmentEpochs) {
+return new RemoteLogSegmentMetadata(
+new RemoteLogSegmentId(new TopicIdPartition(Uuid.randomUuid(),
+new TopicPartition("topic", 0)), Uuid.randomUuid()),
+startOffset, endOffset,
+10L,
+1,
+10L,
+1000,
+Optional.empty(),
+RemoteLogSegmentState.COPY_SEGMENT_FINISHED, segmentEpochs);
+}
+
+@Test
+public void testRemoteSegmentWithinLeaderEpochs() {
+// Test whether a remote segment is within the leader epochs
+final long logEndOffset = 90L;
+
+TreeMap leaderEpochToStartOffset = new TreeMap() {{
+put(0, 0L);
+put(1, 10L);
+put(2, 20L);
+put(3, 30L);
+put(4, 40L);
+put(5, 50L);
+put(7, 70L);
+}};
Review Comment:
For clean code, it creates an anonymous extra class at every usage and we
should try to avoid this pattern.
https://www.baeldung.com/java-initialize-hashmap
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [kafka] kamalcph commented on a diff in pull request #13561: KAFKA-14888: Added remote log segments retention functionality based on time and size.
kamalcph commented on code in PR #13561:
URL: https://github.com/apache/kafka/pull/13561#discussion_r1295394486
##
core/src/main/java/kafka/log/remote/RemoteLogManager.java:
##
@@ -618,6 +625,230 @@ public void run() {
}
}
+public void handleLogStartOffsetUpdate(TopicPartition topicPartition,
long remoteLogStartOffset) {
+if (isLeader()) {
Review Comment:
Filed KAFKA-15351 and KAFKA-15352 to track the cases.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [kafka] kamalcph commented on a diff in pull request #13561: KAFKA-14888: Added remote log segments retention functionality based on time and size.
kamalcph commented on code in PR #13561:
URL: https://github.com/apache/kafka/pull/13561#discussion_r1293158752
##
core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java:
##
@@ -1003,6 +1015,134 @@ public RemoteLogMetadataManager
createRemoteLogMetadataManager() {
}
}
+private static RemoteLogSegmentMetadata
createRemoteLogSegmentMetadata(long startOffset, long endOffset, Map segmentEpochs) {
+return new RemoteLogSegmentMetadata(
+new RemoteLogSegmentId(new TopicIdPartition(Uuid.randomUuid(),
+new TopicPartition("topic", 0)), Uuid.randomUuid()),
+startOffset, endOffset,
+10L,
+1,
+10L,
+1000,
+Optional.empty(),
+RemoteLogSegmentState.COPY_SEGMENT_FINISHED, segmentEpochs);
+}
+
+@Test
+public void testRemoteSegmentWithinLeaderEpochs() {
+// Test whether a remote segment is within the leader epochs
+final long logEndOffset = 90L;
+
+TreeMap leaderEpochToStartOffset = new TreeMap() {{
+put(0, 0L);
+put(1, 10L);
+put(2, 20L);
+put(3, 30L);
+put(4, 40L);
+put(5, 50L);
+put(7, 70L);
+}};
+
+// Test whether a remote segment's epochs/offsets(multiple) are within
the range of leader epochs
+
assertTrue(RemoteLogManager.isRemoteSegmentWithinLeaderEpochs(createRemoteLogSegmentMetadata(
+15,
+35,
+new TreeMap() {{
+put(1, 15L);
+put(2, 20L);
+put(3, 30L);
+}}), logEndOffset, leaderEpochToStartOffset));
+
+// Test whether a remote segment's epochs/offsets(single) are within
the range of leader epochs
+
assertTrue(RemoteLogManager.isRemoteSegmentWithinLeaderEpochs(createRemoteLogSegmentMetadata(
+15,
+19,
+new TreeMap() {{
+put(1, 15L);
+}}), logEndOffset, leaderEpochToStartOffset));
+
+// Test whether a remote segment's epochs/offsets(single) are within
the range of leader epochs
+
assertTrue(RemoteLogManager.isRemoteSegmentWithinLeaderEpochs(createRemoteLogSegmentMetadata(
Review Comment:
Statements in L1065 and L1057 are same. Typo error?
##
core/src/main/java/kafka/log/remote/RemoteLogManager.java:
##
@@ -698,11 +707,329 @@ public void run() {
}
}
+public void handleLogStartOffsetUpdate(TopicPartition topicPartition,
long remoteLogStartOffset) {
+if (isLeader()) {
+logger.debug("Updating {} with remoteLogStartOffset: {}",
topicPartition, remoteLogStartOffset);
+updateRemoteLogStartOffset.accept(topicPartition,
remoteLogStartOffset);
+}
+}
+
+class RemoteLogRetentionHandler {
+
+private final Optional retentionSizeData;
+private final Optional retentionTimeData;
+
+private long remainingBreachedSize;
+
+private OptionalLong logStartOffset = OptionalLong.empty();
+
+public RemoteLogRetentionHandler(Optional
retentionSizeData, Optional retentionTimeData) {
Review Comment:
`Optional` is not recommended as parameter in Java:
https://stackoverflow.com/a/31923105
##
core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java:
##
@@ -1003,6 +1015,134 @@ public RemoteLogMetadataManager
createRemoteLogMetadataManager() {
}
}
+private static RemoteLogSegmentMetadata
createRemoteLogSegmentMetadata(long startOffset, long endOffset, Map segmentEpochs) {
+return new RemoteLogSegmentMetadata(
+new RemoteLogSegmentId(new TopicIdPartition(Uuid.randomUuid(),
+new TopicPartition("topic", 0)), Uuid.randomUuid()),
+startOffset, endOffset,
+10L,
+1,
+10L,
+1000,
+Optional.empty(),
+RemoteLogSegmentState.COPY_SEGMENT_FINISHED, segmentEpochs);
+}
+
+@Test
+public void testRemoteSegmentWithinLeaderEpochs() {
+// Test whether a remote segment is within the leader epochs
+final long logEndOffset = 90L;
+
+TreeMap leaderEpochToStartOffset = new TreeMap() {{
+put(0, 0L);
+put(1, 10L);
+put(2, 20L);
+put(3, 30L);
+put(4, 40L);
+put(5, 50L);
+put(7, 70L);
+}};
Review Comment:
nit:
```suggestion
TreeMap leaderEpochToStartOffset = new TreeMap<>();
leaderEpochToStartOffset.put(0, 0L);
leaderEpochToStartOffset.put(1,
[GitHub] [kafka] kamalcph commented on a diff in pull request #13561: KAFKA-14888: Added remote log segments retention functionality based on time and size.
kamalcph commented on code in PR #13561:
URL: https://github.com/apache/kafka/pull/13561#discussion_r1293081560
##
core/src/main/java/kafka/log/remote/RemoteLogManager.java:
##
@@ -618,6 +625,230 @@ public void run() {
}
}
+public void handleLogStartOffsetUpdate(TopicPartition topicPartition,
long remoteLogStartOffset) {
+if (isLeader()) {
Review Comment:
In the FETCH response, the leader-log-start-offset will be piggy-backed.
But, there can be a scenario:
1. Leader deleted the remote log segment and updates it's log-start-offset
2. Before the replica-2 update it's log-start-offset via FETCH-request, the
leadership changed to replica-2.
3. There are no more eligible segments to delete from remote.
4. The log-start-offset will be stale (referring to old log-start-offset but
the data was already removed from remote)
5. If the consumer starts to read from the beginning of the topic, it will
fail to read.
We can take this task in a follow-up PR if required.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [kafka] kamalcph commented on a diff in pull request #13561: KAFKA-14888: Added remote log segments retention functionality based on time and size.
kamalcph commented on code in PR #13561:
URL: https://github.com/apache/kafka/pull/13561#discussion_r1293081560
##
core/src/main/java/kafka/log/remote/RemoteLogManager.java:
##
@@ -618,6 +625,230 @@ public void run() {
}
}
+public void handleLogStartOffsetUpdate(TopicPartition topicPartition,
long remoteLogStartOffset) {
+if (isLeader()) {
Review Comment:
In the FETCH response, the leader-log-start-offset will be piggy-backed.
But, there can be a scenario:
1. Leader deleted the remote log segment and updates it's log-start-offset
2. Before the replica-2 update it's log-start-offset via FETCH-request, the
leadership changed to replica-2.
3. There are no more eligible segments to delete from remote.
4. The log-start-offset will be stale (referring to old log-start-offset but
the data was already removed from remote)
5. If the consumer starts to read from the beginning of the topic, it will
fail to read.
It's good to handle this case in this PR itself.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
