You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by ja...@apache.org on 2026/09/03 11:20:12 UTC
(fluss) 01/02: [server] Restore KV minRetainOffset on follower restart (#4060)
This is an automated email from the ASF dual-hosted git repository.
wuchong pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
commit 32e36684a82b4d98d82060ba66696cbd98469dd5
Author: Yang Guo <ji...@outlook.com>
AuthorDate: Tue Sep 1 10:12:08 2026 +0800
[server] Restore KV minRetainOffset on follower restart (#4060)
---
.../fluss/rpc/entity/FetchLogResultForBucket.java | 60 +++++++-
.../fluss/rpc/util/CommonRpcMessageUtils.java | 5 +
fluss-rpc/src/main/proto/FlussApi.proto | 3 +
fluss-rust/crates/fluss/proto/FlussApi.proto | 3 +
.../crates/fluss/src/client/table/scanner.rs | 4 +
fluss-rust/crates/fluss/src/proto/fluss.rs | 4 +
.../org/apache/fluss/server/log/LogReadInfo.java | 18 ++-
.../org/apache/fluss/server/log/LogTablet.java | 23 ++--
.../org/apache/fluss/server/replica/Replica.java | 5 +-
.../fluss/server/replica/ReplicaManager.java | 12 +-
.../replica/fetcher/ReplicaFetcherThread.java | 5 +
.../fluss/server/utils/ServerRpcMessageUtils.java | 3 +
.../org/apache/fluss/server/log/LogTabletTest.java | 61 +++++++++
.../replica/KvFollowerRetentionRestoreITCase.java | 151 +++++++++++++++++++++
.../replica/fetcher/ReplicaFetcherThreadTest.java | 114 ++++++++++++++--
.../replica/fetcher/TestingLeaderEndpoint.java | 41 +++++-
.../server/utils/ServerRpcMessageUtilsTest.java | 49 +++++++
17 files changed, 531 insertions(+), 30 deletions(-)
diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/entity/FetchLogResultForBucket.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/entity/FetchLogResultForBucket.java
index e239d8e052..0c5700797a 100644
--- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/entity/FetchLogResultForBucket.java
+++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/entity/FetchLogResultForBucket.java
@@ -27,6 +27,7 @@ import org.apache.fluss.rpc.protocol.ApiError;
import javax.annotation.Nullable;
+import static org.apache.fluss.utils.Preconditions.checkArgument;
import static org.apache.fluss.utils.Preconditions.checkNotNull;
/** Result of {@link FetchLogRequest} for each table bucket. */
@@ -36,6 +37,7 @@ public class FetchLogResultForBucket extends ResultForBucket {
private final @Nullable LogRecords records;
private final long highWatermark;
private final long filteredEndOffset;
+ private final long minRetainOffset;
public FetchLogResultForBucket(
TableBucket tableBucket, LogRecords records, long highWatermark) {
@@ -45,6 +47,7 @@ public class FetchLogResultForBucket extends ResultForBucket {
checkNotNull(records, "records can not be null"),
highWatermark,
-1L,
+ -1L,
ApiError.NONE);
}
@@ -59,11 +62,29 @@ public class FetchLogResultForBucket extends ResultForBucket {
checkNotNull(records, "records can not be null"),
highWatermark,
filteredEndOffset,
+ -1L,
+ ApiError.NONE);
+ }
+
+ /** Creates a successful local fetch result carrying the KV snapshot retention boundary. */
+ public FetchLogResultForBucket(
+ TableBucket tableBucket,
+ LogRecords records,
+ long highWatermark,
+ long filteredEndOffset,
+ long minRetainOffset) {
+ this(
+ tableBucket,
+ null,
+ checkNotNull(records, "records can not be null"),
+ highWatermark,
+ filteredEndOffset,
+ checkMinRetainOffset(minRetainOffset),
ApiError.NONE);
}
public FetchLogResultForBucket(TableBucket tableBucket, ApiError error) {
- this(tableBucket, null, null, -1L, -1L, error);
+ this(tableBucket, null, null, -1L, -1L, -1L, error);
}
public FetchLogResultForBucket(
@@ -74,6 +95,7 @@ public class FetchLogResultForBucket extends ResultForBucket {
null,
highWatermark,
-1L,
+ -1L,
ApiError.NONE);
}
@@ -84,7 +106,7 @@ public class FetchLogResultForBucket extends ResultForBucket {
*/
public FetchLogResultForBucket(
TableBucket tableBucket, long highWatermark, long filteredEndOffset) {
- this(tableBucket, null, null, highWatermark, filteredEndOffset, ApiError.NONE);
+ this(tableBucket, null, null, highWatermark, filteredEndOffset, -1L, ApiError.NONE);
}
private FetchLogResultForBucket(
@@ -93,12 +115,31 @@ public class FetchLogResultForBucket extends ResultForBucket {
@Nullable LogRecords records,
long highWatermark,
long filteredEndOffset,
+ long minRetainOffset,
ApiError error) {
super(tableBucket, error);
this.remoteLogFetchInfo = remoteLogFetchInfo;
this.records = records;
this.highWatermark = highWatermark;
this.filteredEndOffset = filteredEndOffset;
+ this.minRetainOffset = minRetainOffset;
+ }
+
+ /** Returns a successful fetch result carrying the KV snapshot retention boundary. */
+ public FetchLogResultForBucket withMinRetainOffset(long minRetainOffset) {
+ checkArgument(minRetainOffset >= 0, "Min retain offset must be non-negative.");
+ if (failed()) {
+ throw new IllegalStateException(
+ "Min retain offset cannot be attached to a failed fetch result.");
+ }
+ return new FetchLogResultForBucket(
+ getTableBucket(),
+ remoteLogFetchInfo,
+ records,
+ highWatermark,
+ filteredEndOffset,
+ minRetainOffset,
+ ApiError.NONE);
}
/**
@@ -147,4 +188,19 @@ public class FetchLogResultForBucket extends ResultForBucket {
public long getFilteredEndOffset() {
return filteredEndOffset;
}
+
+ /** Returns whether a KV snapshot retention boundary is included in this fetch result. */
+ public boolean hasMinRetainOffset() {
+ return minRetainOffset >= 0;
+ }
+
+ /** Returns the KV snapshot retention boundary included in this fetch result. */
+ public long getMinRetainOffset() {
+ return minRetainOffset;
+ }
+
+ private static long checkMinRetainOffset(long minRetainOffset) {
+ checkArgument(minRetainOffset >= 0, "Min retain offset must be non-negative.");
+ return minRetainOffset;
+ }
}
diff --git a/fluss-rpc/src/main/java/org/apache/fluss/rpc/util/CommonRpcMessageUtils.java b/fluss-rpc/src/main/java/org/apache/fluss/rpc/util/CommonRpcMessageUtils.java
index 219b51087a..bba481fe4a 100644
--- a/fluss-rpc/src/main/java/org/apache/fluss/rpc/util/CommonRpcMessageUtils.java
+++ b/fluss-rpc/src/main/java/org/apache/fluss/rpc/util/CommonRpcMessageUtils.java
@@ -242,6 +242,11 @@ public class CommonRpcMessageUtils {
}
}
+ if (!fetchLogResultForBucket.failed() && respForBucket.hasMinRetainOffset()) {
+ fetchLogResultForBucket =
+ fetchLogResultForBucket.withMinRetainOffset(respForBucket.getMinRetainOffset());
+ }
+
return fetchLogResultForBucket;
}
diff --git a/fluss-rpc/src/main/proto/FlussApi.proto b/fluss-rpc/src/main/proto/FlussApi.proto
index 2a05018d73..e6a1005515 100644
--- a/fluss-rpc/src/main/proto/FlussApi.proto
+++ b/fluss-rpc/src/main/proto/FlussApi.proto
@@ -936,6 +936,9 @@ message PbFetchLogRespForBucket {
// non-empty records field when only trailing batches were filtered. The client should start
// its next fetch from the later of this offset and the end of the records it received.
optional int64 filtered_end_offset = 9;
+ // The safe local log retention boundary confirmed by a committed KV snapshot. This is only
+ // returned for KV follower fetches and is distinct from the physical log_start_offset.
+ optional int64 min_retain_offset = 10;
}
message PbPutKvReqForBucket {
diff --git a/fluss-rust/crates/fluss/proto/FlussApi.proto b/fluss-rust/crates/fluss/proto/FlussApi.proto
index 2a05018d73..e6a1005515 100644
--- a/fluss-rust/crates/fluss/proto/FlussApi.proto
+++ b/fluss-rust/crates/fluss/proto/FlussApi.proto
@@ -936,6 +936,9 @@ message PbFetchLogRespForBucket {
// non-empty records field when only trailing batches were filtered. The client should start
// its next fetch from the later of this offset and the end of the records it received.
optional int64 filtered_end_offset = 9;
+ // The safe local log retention boundary confirmed by a committed KV snapshot. This is only
+ // returned for KV follower fetches and is distinct from the physical log_start_offset.
+ optional int64 min_retain_offset = 10;
}
message PbPutKvReqForBucket {
diff --git a/fluss-rust/crates/fluss/src/client/table/scanner.rs b/fluss-rust/crates/fluss/src/client/table/scanner.rs
index 6aba4c306f..14c587e57d 100644
--- a/fluss-rust/crates/fluss/src/client/table/scanner.rs
+++ b/fluss-rust/crates/fluss/src/client/table/scanner.rs
@@ -2998,6 +2998,7 @@ mod tests {
remote_log_fetch_info: None,
records: None,
filtered_end_offset,
+ min_retain_offset: None,
}],
}],
}
@@ -3052,6 +3053,7 @@ mod tests {
remote_log_fetch_info: None,
records: None,
filtered_end_offset: None,
+ min_retain_offset: None,
}],
}],
};
@@ -3111,6 +3113,7 @@ mod tests {
remote_log_fetch_info: None,
records: None,
filtered_end_offset: None,
+ min_retain_offset: None,
}],
}],
};
@@ -3460,6 +3463,7 @@ mod tests {
remote_log_fetch_info: None,
records: None,
filtered_end_offset: None,
+ min_retain_offset: None,
}],
}],
};
diff --git a/fluss-rust/crates/fluss/src/proto/fluss.rs b/fluss-rust/crates/fluss/src/proto/fluss.rs
index 3db6ada950..99c5a6ffe9 100644
--- a/fluss-rust/crates/fluss/src/proto/fluss.rs
+++ b/fluss-rust/crates/fluss/src/proto/fluss.rs
@@ -1245,6 +1245,10 @@ pub struct PbFetchLogRespForBucket {
/// its next fetch from the later of this offset and the end of the records it received.
#[prost(int64, optional, tag = "9")]
pub filtered_end_offset: ::core::option::Option<i64>,
+ /// The safe local log retention boundary confirmed by a committed KV snapshot. This is only
+ /// returned for KV follower fetches and is distinct from the physical log_start_offset.
+ #[prost(int64, optional, tag = "10")]
+ pub min_retain_offset: ::core::option::Option<i64>,
}
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
pub struct PbPutKvReqForBucket {
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/log/LogReadInfo.java b/fluss-server/src/main/java/org/apache/fluss/server/log/LogReadInfo.java
index 150c3c38ff..0d14fb2412 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/log/LogReadInfo.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/log/LogReadInfo.java
@@ -26,11 +26,17 @@ public class LogReadInfo {
private final FetchDataInfo fetchedData;
private final long highWatermark;
private final long logEndOffset;
+ private final long minRetainOffset;
- public LogReadInfo(FetchDataInfo fetchedData, long highWatermark, long logEndOffset) {
+ public LogReadInfo(
+ FetchDataInfo fetchedData,
+ long highWatermark,
+ long logEndOffset,
+ long minRetainOffset) {
this.fetchedData = fetchedData;
this.highWatermark = highWatermark;
this.logEndOffset = logEndOffset;
+ this.minRetainOffset = minRetainOffset;
}
public FetchDataInfo getFetchedData() {
@@ -45,6 +51,14 @@ public class LogReadInfo {
return logEndOffset;
}
+ public boolean hasMinRetainOffset() {
+ return minRetainOffset >= 0;
+ }
+
+ public long getMinRetainOffset() {
+ return minRetainOffset;
+ }
+
@Override
public String toString() {
return "LogReadInfo("
@@ -54,6 +68,8 @@ public class LogReadInfo {
+ highWatermark
+ ", logEndOffset="
+ logEndOffset
+ + ", minRetainOffset="
+ + minRetainOffset
+ ')';
}
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/log/LogTablet.java b/fluss-server/src/main/java/org/apache/fluss/server/log/LogTablet.java
index 610ed4bf24..fa45ae34d2 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/log/LogTablet.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/log/LogTablet.java
@@ -66,6 +66,7 @@ import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
import static org.apache.fluss.utils.Preconditions.checkArgument;
import static org.apache.fluss.utils.Preconditions.checkNotNull;
@@ -123,7 +124,7 @@ public final class LogTablet {
// The minimum offset that should be retained in the local log. This is used to ensure that,
// the offset of kv snapshot should be retained, otherwise, kv recovery will fail.
- private volatile long minRetainOffset;
+ private final AtomicLong minRetainOffset;
// tracking the log start offset in remote storage
private volatile long remoteLogStartOffset = Long.MAX_VALUE;
// tracking the log end offset in remote storage
@@ -185,7 +186,7 @@ public final class LogTablet {
// Default value to 0L for changelog to avoid cleaning up any segments in case of not
// updating this value in time. Default value to Long.MAX_VALUE for normal log table,
// as we don't need to retain logs for kv recovery.
- this.minRetainOffset = isChangelog ? 0L : Long.MAX_VALUE;
+ this.minRetainOffset = new AtomicLong(isChangelog ? 0L : Long.MAX_VALUE);
}
public PhysicalTablePath getPhysicalTablePath() {
@@ -672,11 +673,14 @@ public final class LogTablet {
}
public void updateMinRetainOffset(long minRetainOffset) {
- if (minRetainOffset > this.minRetainOffset) {
- this.minRetainOffset = minRetainOffset;
-
- // try to delete the old segments that are not needed.
- deleteSegmentsAlreadyExistsInRemote();
+ long currentMinRetainOffset = this.minRetainOffset.get();
+ while (minRetainOffset > currentMinRetainOffset) {
+ if (this.minRetainOffset.compareAndSet(currentMinRetainOffset, minRetainOffset)) {
+ // try to delete the old segments that are not needed.
+ deleteSegmentsAlreadyExistsInRemote();
+ return;
+ }
+ currentMinRetainOffset = this.minRetainOffset.get();
}
}
@@ -830,7 +834,8 @@ public final class LogTablet {
try {
// shouldn't clean up segments that will be used by kv recovery.
- long effectiveCleanupToOffset = Math.min(minRetainOffset, requestedCleanupToOffset);
+ long effectiveCleanupToOffset =
+ Math.min(minRetainOffset.get(), requestedCleanupToOffset);
cleanupAction.cleanup(effectiveCleanupToOffset);
} catch (IOException e) {
LOG.error(
@@ -1611,6 +1616,6 @@ public final class LogTablet {
@VisibleForTesting
public long getMinRetainOffset() {
- return minRetainOffset;
+ return minRetainOffset.get();
}
}
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java
index ee28567756..bb6489213e 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java
@@ -1921,6 +1921,8 @@ public final class Replica {
// the fetch do not prevent a follower from coming into sync.
long initialHighWatermark = logTablet.getHighWatermark();
long initialLogEndOffset = logTablet.localLogEndOffset();
+ long initialMinRetainOffset =
+ fetchParams.isFromFollower() && isKvTable() ? logTablet.getMinRetainOffset() : -1L;
long readOffset =
fetchParams.fetchOffset() == FetchParams.FETCH_FROM_EARLIEST_OFFSET
? logTablet.logStartOffset()
@@ -1947,7 +1949,8 @@ public final class Replica {
IOUtils.closeQuietly(filterContext.getReadContext());
}
}
- return new LogReadInfo(fetchDataInfo, initialHighWatermark, initialLogEndOffset);
+ return new LogReadInfo(
+ fetchDataInfo, initialHighWatermark, initialLogEndOffset, initialMinRetainOffset);
}
/**
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java
index b335701923..5e8c0810df 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java
@@ -1787,7 +1787,17 @@ public class ReplicaManager implements ServerReconfigurable {
}
limitBytes = Math.max(0, limitBytes - recordBatchSize);
FetchLogResultForBucket fetchLogResult;
- if (fetchedData.hasFilteredEndOffset()) {
+ if (readInfo.hasMinRetainOffset()) {
+ fetchLogResult =
+ new FetchLogResultForBucket(
+ tb,
+ fetchedData.getRecords(),
+ readInfo.getHighWatermark(),
+ fetchedData.hasFilteredEndOffset()
+ ? fetchedData.getFilteredEndOffset()
+ : -1L,
+ readInfo.getMinRetainOffset());
+ } else if (fetchedData.hasFilteredEndOffset()) {
fetchLogResult =
new FetchLogResultForBucket(
tb,
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherThread.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherThread.java
index cc430a19f9..c9c7ced85c 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherThread.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherThread.java
@@ -344,6 +344,11 @@ final class ReplicaFetcherThread extends ShutdownableThread {
}
}
+ Replica replica = replicaManager.getReplicaOrException(tableBucket);
+ if (replica.isKvTable() && replicaData.hasMinRetainOffset()) {
+ replica.getLogTablet().updateMinRetainOffset(replicaData.getMinRetainOffset());
+ }
+
if (nextFetchOffset != -1L && fairBucketStatusMap.contains(tableBucket)) {
BucketFetchStatus newFetchStatus =
new BucketFetchStatus(
diff --git a/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java b/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java
index 9952d48571..c66d4a7c6d 100644
--- a/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java
+++ b/fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java
@@ -1024,6 +1024,9 @@ public class ServerRpcMessageUtils {
if (bucketResult.hasFilteredEndOffset()) {
fetchLogRespForBucket.setFilteredEndOffset(bucketResult.getFilteredEndOffset());
}
+ if (bucketResult.hasMinRetainOffset()) {
+ fetchLogRespForBucket.setMinRetainOffset(bucketResult.getMinRetainOffset());
+ }
if (tb.getPartitionId() != null) {
fetchLogRespForBucket.setPartitionId(tb.getPartitionId());
}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/log/LogTabletTest.java b/fluss-server/src/test/java/org/apache/fluss/server/log/LogTabletTest.java
index aae7ff3d00..691a7af479 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/log/LogTabletTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/log/LogTabletTest.java
@@ -51,7 +51,11 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import static org.apache.fluss.record.TestData.DATA1;
@@ -124,6 +128,63 @@ final class LogTabletTest extends LogTestBase {
assertThat(logTablet.canFetchFromRemoteLog(0L)).isTrue();
}
+ @Test
+ void testMinRetainOffsetIsMonotonicWithConcurrentUpdates() throws Exception {
+ File kvLogDir =
+ LogTestUtils.makeRandomLogTabletDir(
+ tempDir,
+ DATA1_TABLE_PATH.getDatabaseName(),
+ DATA1_TABLE_ID,
+ DATA1_TABLE_PATH.getTableName());
+ LogTablet kvLogTablet =
+ LogTablet.create(
+ tempDir,
+ PhysicalTablePath.of(DATA1_TABLE_PATH),
+ kvLogDir,
+ conf,
+ new AtomicBoolean(false),
+ TestingMetricGroups.TABLET_SERVER_METRICS,
+ 0,
+ scheduler,
+ LogFormat.ARROW,
+ 1,
+ true,
+ SystemClock.getInstance(),
+ true);
+ int updateThreadCount = 32;
+ ExecutorService executor = Executors.newFixedThreadPool(updateThreadCount);
+ CountDownLatch ready = new CountDownLatch(updateThreadCount);
+ CountDownLatch start = new CountDownLatch(1);
+ try {
+ for (int i = 1; i <= updateThreadCount; i++) {
+ final long minRetainOffset = i;
+ executor.execute(
+ () -> {
+ ready.countDown();
+ try {
+ start.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return;
+ }
+ kvLogTablet.updateMinRetainOffset(minRetainOffset);
+ });
+ }
+
+ assertThat(ready.await(10, TimeUnit.SECONDS)).isTrue();
+ start.countDown();
+ executor.shutdown();
+ assertThat(executor.awaitTermination(10, TimeUnit.SECONDS)).isTrue();
+
+ assertThat(kvLogTablet.getMinRetainOffset()).isEqualTo(updateThreadCount);
+ kvLogTablet.updateMinRetainOffset(1L);
+ assertThat(kvLogTablet.getMinRetainOffset()).isEqualTo(updateThreadCount);
+ } finally {
+ executor.shutdownNow();
+ kvLogTablet.close();
+ }
+ }
+
@Test
void testHighWatermarkMetadataUpdatedAfterSegmentRoll() throws Exception {
MemoryLogRecords mr =
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/KvFollowerRetentionRestoreITCase.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/KvFollowerRetentionRestoreITCase.java
new file mode 100644
index 0000000000..3bb45bb764
--- /dev/null
+++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/KvFollowerRetentionRestoreITCase.java
@@ -0,0 +1,151 @@
+/*
+ * 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.fluss.server.replica;
+
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.config.MemorySize;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.metadata.TableDescriptor;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.record.KvRecord;
+import org.apache.fluss.rpc.gateway.TabletServerGateway;
+import org.apache.fluss.rpc.messages.PutKvResponse;
+import org.apache.fluss.server.kv.snapshot.CompletedSnapshot;
+import org.apache.fluss.server.testutils.FlussClusterExtension;
+import org.apache.fluss.server.zk.data.LeaderAndIsr;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.fluss.record.TestData.DATA1_SCHEMA_PK;
+import static org.apache.fluss.server.testutils.RpcMessageTestUtils.createTable;
+import static org.apache.fluss.server.testutils.RpcMessageTestUtils.newPutKvRequest;
+import static org.apache.fluss.testutils.DataTestUtils.genKvRecords;
+import static org.apache.fluss.testutils.DataTestUtils.toKvRecordBatch;
+import static org.apache.fluss.testutils.common.CommonTestUtils.retry;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** IT case for restoring a follower's KV log retention boundary after restart. */
+class KvFollowerRetentionRestoreITCase {
+
+ @RegisterExtension
+ public static final FlussClusterExtension FLUSS_CLUSTER_EXTENSION =
+ FlussClusterExtension.builder()
+ .setNumOfTabletServers(2)
+ .setClusterConf(initConfig())
+ .build();
+
+ @Test
+ void testCleanupLocalLogsAfterFollowerRestartWithoutNewWrites() throws Exception {
+ TablePath tablePath = TablePath.of("test_db", "test_follower_retention_restore");
+ TableDescriptor tableDescriptor =
+ TableDescriptor.builder()
+ .schema(DATA1_SCHEMA_PK)
+ .distributedBy(1, "a")
+ .property(ConfigOptions.TABLE_LOG_TTL, Duration.ofMillis(1))
+ .build();
+ long tableId = createTable(FLUSS_CLUSTER_EXTENSION, tablePath, tableDescriptor);
+ TableBucket tableBucket = new TableBucket(tableId, 0);
+ FLUSS_CLUSTER_EXTENSION.waitUntilAllReplicaReady(tableBucket);
+
+ LeaderAndIsr leaderAndIsr = FLUSS_CLUSTER_EXTENSION.waitLeaderAndIsrReady(tableBucket);
+ int leader = leaderAndIsr.leader();
+ int follower =
+ leaderAndIsr.isr().stream().filter(replica -> replica != leader).findFirst().get();
+ TabletServerGateway leaderGateway =
+ FLUSS_CLUSTER_EXTENSION.newTabletServerClientForNode(leader);
+
+ for (int batchId = 0; batchId < 4; batchId++) {
+ List<KvRecord> records = new ArrayList<>();
+ for (int recordId = 0; recordId < 10; recordId++) {
+ int key = batchId * 10 + recordId;
+ records.addAll(genKvRecords(new Object[] {key, "value_" + key}));
+ }
+ PutKvResponse response =
+ leaderGateway
+ .putKv(
+ newPutKvRequest(
+ tableId,
+ tableBucket.getBucket(),
+ -1,
+ toKvRecordBatch(records)))
+ .get();
+ assertThat(response.getBucketsRespAt(0).hasErrorCode()).isFalse();
+ }
+
+ Replica leaderReplica = FLUSS_CLUSTER_EXTENSION.waitAndGetLeaderReplica(tableBucket);
+ Replica followerReplica =
+ FLUSS_CLUSTER_EXTENSION.waitAndGetFollowerReplica(tableBucket, follower);
+ long logEndOffset = leaderReplica.getLocalLogEndOffset();
+ retry(
+ Duration.ofMinutes(1),
+ () -> {
+ assertThat(followerReplica.getLocalLogEndOffset()).isEqualTo(logEndOffset);
+ assertThat(followerReplica.getLogTablet().getHighWatermark())
+ .isEqualTo(logEndOffset);
+ assertThat(followerReplica.getLogTablet().getSegments().size())
+ .isGreaterThan(1);
+ });
+ int segmentsBeforeRestart = followerReplica.getLogTablet().getSegments().size();
+
+ // The stopped follower misses the snapshot notification that advances minRetainOffset.
+ FLUSS_CLUSTER_EXTENSION.stopTabletServer(follower);
+ FLUSS_CLUSTER_EXTENSION.waitUntilReplicaShrinkFromIsr(tableBucket, follower);
+ CompletedSnapshot snapshot = FLUSS_CLUSTER_EXTENSION.triggerAndWaitSnapshot(tableBucket);
+ assertThat(snapshot.getLogOffset()).isEqualTo(logEndOffset);
+ retry(
+ Duration.ofMinutes(1),
+ () ->
+ assertThat(leaderReplica.getLogTablet().getMinRetainOffset())
+ .isEqualTo(logEndOffset));
+
+ // Do not write after the snapshot. The restarted follower must recover the retention
+ // boundary from a successful empty fetch response.
+ FLUSS_CLUSTER_EXTENSION.startTabletServer(follower);
+ FLUSS_CLUSTER_EXTENSION.waitUntilReplicaExpandToIsr(tableBucket, follower);
+ Replica restartedFollower =
+ FLUSS_CLUSTER_EXTENSION.waitAndGetFollowerReplica(tableBucket, follower);
+ retry(
+ Duration.ofMinutes(1),
+ () -> {
+ assertThat(restartedFollower.getLocalLogEndOffset()).isEqualTo(logEndOffset);
+ assertThat(restartedFollower.getLogTablet().getMinRetainOffset())
+ .isEqualTo(logEndOffset);
+ assertThat(restartedFollower.getLogTablet().getSegments().size())
+ .isLessThan(segmentsBeforeRestart);
+ assertThat(restartedFollower.getLocalLogStartOffset()).isPositive();
+ });
+ assertThat(leaderReplica.getLocalLogEndOffset()).isEqualTo(logEndOffset);
+ }
+
+ private static Configuration initConfig() {
+ Configuration conf = new Configuration();
+ conf.setInt(ConfigOptions.DEFAULT_REPLICATION_FACTOR, 2);
+ conf.set(ConfigOptions.LOG_SEGMENT_FILE_SIZE, MemorySize.parse("100b"));
+ conf.set(ConfigOptions.LOG_RETENTION_CHECK_INTERVAL, Duration.ofMillis(100));
+ conf.set(ConfigOptions.REMOTE_LOG_TASK_INTERVAL_DURATION, Duration.ZERO);
+ conf.set(ConfigOptions.KV_SNAPSHOT_INTERVAL, Duration.ofHours(1));
+ conf.set(ConfigOptions.LOG_REPLICA_MAX_LAG_TIME, Duration.ofSeconds(5));
+ return conf;
+ }
+}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherThreadTest.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherThreadTest.java
index 8a1f6fb41e..37b764c0bf 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherThreadTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/fetcher/ReplicaFetcherThreadTest.java
@@ -24,6 +24,7 @@ import org.apache.fluss.config.Configuration;
import org.apache.fluss.exception.FlussRuntimeException;
import org.apache.fluss.metadata.PhysicalTablePath;
import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.record.MemoryLogRecords;
import org.apache.fluss.rpc.RpcClient;
import org.apache.fluss.rpc.entity.ProduceLogResultForBucket;
import org.apache.fluss.rpc.metrics.TestingClientMetricGroup;
@@ -76,9 +77,13 @@ import java.util.function.Consumer;
import static org.apache.fluss.record.TestData.DATA1;
import static org.apache.fluss.record.TestData.DATA1_SCHEMA;
+import static org.apache.fluss.record.TestData.DATA1_SCHEMA_PK;
import static org.apache.fluss.record.TestData.DATA1_TABLE_DESCRIPTOR;
+import static org.apache.fluss.record.TestData.DATA1_TABLE_DESCRIPTOR_PK;
import static org.apache.fluss.record.TestData.DATA1_TABLE_ID;
+import static org.apache.fluss.record.TestData.DATA1_TABLE_ID_PK;
import static org.apache.fluss.record.TestData.DATA1_TABLE_PATH;
+import static org.apache.fluss.record.TestData.DATA1_TABLE_PATH_PK;
import static org.apache.fluss.record.TestData.DEFAULT_REMOTE_DATA_DIR;
import static org.apache.fluss.server.coordinator.CoordinatorContext.INITIAL_COORDINATOR_EPOCH;
import static org.apache.fluss.server.metrics.group.TestingMetricGroups.USER_METRICS;
@@ -105,9 +110,12 @@ public class ReplicaFetcherThreadTest {
private ServerNode leader;
private ReplicaManager followerRM;
private ReplicaFetcherThread followerFetcher;
+ private TestingLeaderEndpoint leaderEndpoint;
private ExecutorService ioExecutor;
private LocalDiskManager leaderLocalDiskManager;
private LocalDiskManager followerLocalDiskManager;
+ private KvManager leaderKvManager;
+ private KvManager followerKvManager;
@BeforeAll
static void baseBeforeAll() {
@@ -122,6 +130,7 @@ public class ReplicaFetcherThreadTest {
ZOO_KEEPER_EXTENSION_WRAPPER.getCustomExtension().cleanupRoot();
manualClock = new ManualClock(System.currentTimeMillis());
Configuration conf = new Configuration();
+ conf.set(ConfigOptions.LOG_REPLICA_FETCH_WAIT_MAX_TIME, Duration.ofSeconds(5));
tb = new TableBucket(DATA1_TABLE_ID, 0);
ioExecutor = Executors.newSingleThreadExecutor();
leaderLocalDiskManager = createLocalDiskManager(leaderServerId);
@@ -135,12 +144,9 @@ public class ReplicaFetcherThreadTest {
ServerNode follower =
new ServerNode(
followerServerId, "localhost", 10001, ServerType.TABLET_SERVER, "rack2");
+ leaderEndpoint = new TestingLeaderEndpoint(conf, leaderRM, follower);
followerFetcher =
- new ReplicaFetcherThread(
- "test-fetcher-thread",
- followerRM,
- new TestingLeaderEndpoint(conf, leaderRM, follower),
- 1000);
+ new ReplicaFetcherThread("test-fetcher-thread", followerRM, leaderEndpoint, 1000);
registerTableInZkClient();
// make the tb(table, 0) to be leader in leaderRM and to be follower in followerRM.
@@ -149,6 +155,21 @@ public class ReplicaFetcherThreadTest {
@AfterEach
public void tearDown() throws Exception {
+ if (followerFetcher != null && followerFetcher.isAlive()) {
+ followerFetcher.shutdown();
+ }
+ if (leaderRM != null) {
+ leaderRM.getDelayedFetchLogManager().shutdown();
+ }
+ if (followerRM != null) {
+ followerRM.getDelayedFetchLogManager().shutdown();
+ }
+ if (leaderKvManager != null) {
+ leaderKvManager.shutdown();
+ }
+ if (followerKvManager != null) {
+ followerKvManager.shutdown();
+ }
if (leaderLocalDiskManager != null) {
leaderLocalDiskManager.close();
}
@@ -202,6 +223,59 @@ public class ReplicaFetcherThreadTest {
.isEqualTo(20L));
}
+ @Test
+ void testRestoreKvMinRetainOffsetFromDelayedEmptyFetchResponse() throws Exception {
+ leaderEndpoint.enableLongPolling();
+ TableBucket pkTableBucket = new TableBucket(DATA1_TABLE_ID_PK, 0);
+ zkClient.registerTable(
+ DATA1_TABLE_PATH_PK,
+ TableRegistration.newTable(
+ DATA1_TABLE_ID_PK, DEFAULT_REMOTE_DATA_DIR, DATA1_TABLE_DESCRIPTOR_PK));
+ zkClient.registerFirstSchema(DATA1_TABLE_PATH_PK, DATA1_SCHEMA_PK);
+ makeLeaderAndFollower(PhysicalTablePath.of(DATA1_TABLE_PATH_PK), pkTableBucket);
+
+ Replica leaderReplica = leaderRM.getReplicaOrException(pkTableBucket);
+ Replica followerReplica = followerRM.getReplicaOrException(pkTableBucket);
+ for (int i = 0; i < 3; i++) {
+ MemoryLogRecords records = genMemoryLogRecordsWithWriterId(DATA1, 100L + i, 0, i * 10L);
+ leaderReplica.appendRecordsToFollower(records);
+ followerReplica.appendRecordsToFollower(records);
+ if (i < 2) {
+ leaderReplica.getLogTablet().roll(Optional.empty());
+ followerReplica.getLogTablet().roll(Optional.empty());
+ }
+ }
+ leaderReplica.getLogTablet().updateHighWatermark(30L);
+ followerReplica.getLogTablet().updateHighWatermark(30L);
+ leaderReplica.getLogTablet().updateMinRetainOffset(30L);
+ followerReplica.getLogTablet().updateHighestCopiedEndOffset(30L);
+
+ assertThat(leaderReplica.getLocalLogEndOffset()).isEqualTo(30L);
+ assertThat(followerReplica.getLocalLogEndOffset()).isEqualTo(30L);
+ assertThat(followerReplica.getLogTablet().getMinRetainOffset()).isZero();
+ assertThat(followerReplica.getLogTablet().getSegments()).hasSize(3);
+
+ followerFetcher.addBuckets(
+ Collections.singletonMap(
+ pkTableBucket,
+ new InitialFetchStatus(
+ DATA1_TABLE_ID_PK, DATA1_TABLE_PATH_PK, leader.id(), 30L)));
+ followerFetcher.start();
+
+ retry(
+ Duration.ofSeconds(20),
+ () -> assertThat(leaderRM.getDelayedFetchLogManager().numDelayed()).isPositive());
+ assertThat(followerReplica.getLogTablet().getMinRetainOffset()).isZero();
+ assertThat(followerReplica.getLogTablet().getSegments()).hasSize(3);
+
+ retry(
+ Duration.ofSeconds(20),
+ () -> {
+ assertThat(followerReplica.getLogTablet().getMinRetainOffset()).isEqualTo(30L);
+ assertThat(followerReplica.getLogTablet().getSegments()).hasSize(2);
+ });
+ }
+
@Test
void testFollowerHighWatermarkHigherThanOrEqualToLeader() throws Exception {
Replica leaderReplica = leaderRM.getReplicaOrException(tb);
@@ -478,12 +552,17 @@ public class ReplicaFetcherThreadTest {
}
private void makeLeaderAndFollower() {
+ makeLeaderAndFollower(PhysicalTablePath.of(DATA1_TABLE_PATH), tb);
+ }
+
+ private void makeLeaderAndFollower(
+ PhysicalTablePath physicalTablePath, TableBucket tableBucket) {
leaderRM.becomeLeaderOrFollower(
INITIAL_COORDINATOR_EPOCH,
Collections.singletonList(
new NotifyLeaderAndIsrData(
- PhysicalTablePath.of(DATA1_TABLE_PATH),
- tb,
+ physicalTablePath,
+ tableBucket,
Arrays.asList(leaderServerId, followerServerId),
new LeaderAndIsr(
leaderServerId,
@@ -497,8 +576,8 @@ public class ReplicaFetcherThreadTest {
INITIAL_COORDINATOR_EPOCH,
Collections.singletonList(
new NotifyLeaderAndIsrData(
- PhysicalTablePath.of(DATA1_TABLE_PATH),
- tb,
+ physicalTablePath,
+ tableBucket,
Arrays.asList(leaderServerId, followerServerId),
new LeaderAndIsr(
leaderServerId,
@@ -535,12 +614,27 @@ public class ReplicaFetcherThreadTest {
TestingMetricGroups.TABLET_SERVER_METRICS,
localDiskManager);
logManager.startup();
+ KvManager kvManager =
+ KvManager.create(
+ conf,
+ zkClient,
+ logManager,
+ TestingMetricGroups.TABLET_SERVER_METRICS,
+ localDiskManager,
+ null,
+ manualClock);
+ kvManager.startup();
+ if (serverId == leaderServerId) {
+ leaderKvManager = kvManager;
+ } else {
+ followerKvManager = kvManager;
+ }
ReplicaManager replicaManager =
new TestingReplicaManager(
conf,
scheduler,
logManager,
- null,
+ kvManager,
zkClient,
serverId,
new TabletServerMetadataCache(
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/fetcher/TestingLeaderEndpoint.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/fetcher/TestingLeaderEndpoint.java
index b2ffd28926..fcf6f00b5e 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/replica/fetcher/TestingLeaderEndpoint.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/fetcher/TestingLeaderEndpoint.java
@@ -55,6 +55,10 @@ public class TestingLeaderEndpoint implements LeaderEndpoint {
/** The max fetch size for a bucket in bytes. */
private final int maxFetchSizeForBucket;
+ private final int minFetchBytes;
+ private final int maxFetchWaitMs;
+ private volatile boolean longPollingEnabled;
+
/**
* If set, fetchLog will return a future that completes after the specified delay, simulating a
* slow leader. The response will carry a pooled ByteBuf to track buffer release.
@@ -74,6 +78,9 @@ public class TestingLeaderEndpoint implements LeaderEndpoint {
this.maxFetchSize = (int) conf.get(ConfigOptions.LOG_REPLICA_FETCH_MAX_BYTES).getBytes();
this.maxFetchSizeForBucket =
(int) conf.get(ConfigOptions.LOG_REPLICA_FETCH_MAX_BYTES_FOR_BUCKET).getBytes();
+ this.minFetchBytes = (int) conf.get(ConfigOptions.LOG_REPLICA_FETCH_MIN_BYTES).getBytes();
+ this.maxFetchWaitMs =
+ (int) conf.get(ConfigOptions.LOG_REPLICA_FETCH_WAIT_MAX_TIME).toMillis();
}
@Override
@@ -106,7 +113,10 @@ public class TestingLeaderEndpoint implements LeaderEndpoint {
Map<TableBucket, FetchReqInfo> fetchLogData = getFetchLogData(fetchLogRequest);
replicaManager.fetchLogRecords(
new FetchParams(
- fetchLogRequest.getFollowerServerId(), fetchLogRequest.getMaxBytes()),
+ fetchLogRequest.getFollowerServerId(),
+ fetchLogRequest.getMaxBytes(),
+ fetchLogRequest.getMinBytes(),
+ fetchLogRequest.getMaxWaitMs()),
fetchLogData,
null,
result -> {
@@ -139,6 +149,11 @@ public class TestingLeaderEndpoint implements LeaderEndpoint {
this.delayMs = 0;
}
+ /** Enables the configured follower fetch long-poll settings for tests. */
+ public void enableLongPolling() {
+ longPollingEnabled = true;
+ }
+
/** Returns all pooled ByteBufs allocated for delayed fetch responses. */
public java.util.List<ByteBuf> getAllAllocatedByteBufs() {
return allocatedByteBufs;
@@ -166,7 +181,12 @@ public class TestingLeaderEndpoint implements LeaderEndpoint {
public Optional<FetchLogContext> buildFetchLogContext(
Map<TableBucket, BucketFetchStatus> replicas) {
return RemoteLeaderEndpoint.buildFetchLogContext(
- replicas, localNode.id(), maxFetchSize, maxFetchSizeForBucket, -1, -1);
+ replicas,
+ localNode.id(),
+ maxFetchSize,
+ maxFetchSizeForBucket,
+ longPollingEnabled ? minFetchBytes : FetchParams.DEFAULT_MIN_FETCH_BYTES,
+ longPollingEnabled ? maxFetchWaitMs : (int) FetchParams.DEFAULT_MAX_WAIT_MS);
}
@Override
@@ -187,10 +207,19 @@ public class TestingLeaderEndpoint implements LeaderEndpoint {
ByteBuffer buffer = ByteBuffer.allocate(fileRecords.sizeInBytes());
unchecked(() -> fileRecords.readInto(buffer, 0)).run();
MemoryLogRecords memRecords = MemoryLogRecords.pointToByteBuffer(buffer);
- result.put(
- tb,
- new FetchLogResultForBucket(
- tb, memRecords, value.getHighWatermark()));
+ FetchLogResultForBucket memoryResult =
+ value.hasMinRetainOffset()
+ ? new FetchLogResultForBucket(
+ tb,
+ memRecords,
+ value.getHighWatermark(),
+ value.hasFilteredEndOffset()
+ ? value.getFilteredEndOffset()
+ : -1L,
+ value.getMinRetainOffset())
+ : new FetchLogResultForBucket(
+ tb, memRecords, value.getHighWatermark());
+ result.put(tb, memoryResult);
} else {
result.put(tb, value);
}
diff --git a/fluss-server/src/test/java/org/apache/fluss/server/utils/ServerRpcMessageUtilsTest.java b/fluss-server/src/test/java/org/apache/fluss/server/utils/ServerRpcMessageUtilsTest.java
index 1ed9660e51..00c187fca4 100644
--- a/fluss-server/src/test/java/org/apache/fluss/server/utils/ServerRpcMessageUtilsTest.java
+++ b/fluss-server/src/test/java/org/apache/fluss/server/utils/ServerRpcMessageUtilsTest.java
@@ -18,11 +18,16 @@
package org.apache.fluss.server.utils;
import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.metadata.TablePath;
import org.apache.fluss.record.KvRecordBatch;
+import org.apache.fluss.record.MemoryLogRecords;
import org.apache.fluss.row.encode.KvValueLayout;
+import org.apache.fluss.rpc.entity.FetchLogResultForBucket;
import org.apache.fluss.rpc.entity.LookupResultForBucket;
import org.apache.fluss.rpc.entity.PutKvResultForBucket;
+import org.apache.fluss.rpc.messages.FetchLogResponse;
import org.apache.fluss.rpc.messages.LookupResponse;
+import org.apache.fluss.rpc.messages.PbFetchLogRespForBucket;
import org.apache.fluss.rpc.messages.PbBucketMetadata;
import org.apache.fluss.rpc.messages.PbPartitionMetadata;
import org.apache.fluss.rpc.messages.PbPutKvReqForBucket;
@@ -30,6 +35,7 @@ import org.apache.fluss.rpc.messages.PbPutKvRespForBucket;
import org.apache.fluss.rpc.messages.PbTableMetadata;
import org.apache.fluss.rpc.messages.PutKvRequest;
import org.apache.fluss.rpc.messages.PutKvResponse;
+import org.apache.fluss.rpc.util.CommonRpcMessageUtils;
import org.apache.fluss.rpc.messages.UpdateMetadataRequest;
import org.apache.fluss.server.entity.PutKvDataForBucket;
import org.apache.fluss.server.metadata.BucketMetadata;
@@ -56,6 +62,49 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link ServerRpcMessageUtils}. */
class ServerRpcMessageUtilsTest {
+ @Test
+ void testFetchLogResponseContainsMinRetainOffset() {
+ TableBucket tableBucket = new TableBucket(1L, 0);
+ FetchLogResultForBucket result =
+ new FetchLogResultForBucket(tableBucket, MemoryLogRecords.EMPTY, 10L, -1L, 8L);
+
+ FetchLogResponse response =
+ ServerRpcMessageUtils.makeFetchLogResponse(
+ Collections.singletonMap(tableBucket, result));
+
+ PbFetchLogRespForBucket bucketResponse =
+ response.getTablesRespsList().get(0).getBucketsRespsList().get(0);
+ assertThat(bucketResponse.hasMinRetainOffset()).isTrue();
+ assertThat(bucketResponse.getMinRetainOffset()).isEqualTo(8L);
+
+ FetchLogResultForBucket decodedResult =
+ CommonRpcMessageUtils.getFetchLogResultForBucket(
+ tableBucket, TablePath.of("db", "table"), bucketResponse);
+ assertThat(decodedResult.hasMinRetainOffset()).isTrue();
+ assertThat(decodedResult.getMinRetainOffset()).isEqualTo(8L);
+
+ FetchLogResponse responseWithoutMinRetainOffset =
+ ServerRpcMessageUtils.makeFetchLogResponse(
+ Collections.singletonMap(
+ tableBucket,
+ new FetchLogResultForBucket(
+ tableBucket, MemoryLogRecords.EMPTY, 10L)));
+ PbFetchLogRespForBucket bucketResponseWithoutMinRetainOffset =
+ responseWithoutMinRetainOffset
+ .getTablesRespsList()
+ .get(0)
+ .getBucketsRespsList()
+ .get(0);
+ assertThat(bucketResponseWithoutMinRetainOffset.hasMinRetainOffset()).isFalse();
+ assertThat(
+ CommonRpcMessageUtils.getFetchLogResultForBucket(
+ tableBucket,
+ TablePath.of("db", "table"),
+ bucketResponseWithoutMinRetainOffset)
+ .hasMinRetainOffset())
+ .isFalse();
+ }
+
@Test
void testLookupResponseSupportsMixedValueLayouts() {
byte[] plainValue = new byte[] {1, 0, 11, 12};