diff --git a/fluss-client/src/main/java/com/alibaba/fluss/client/metadata/MetadataUpdater.java b/fluss-client/src/main/java/com/alibaba/fluss/client/metadata/MetadataUpdater.java index 6fb12669c9..82d7bd4075 100644 --- a/fluss-client/src/main/java/com/alibaba/fluss/client/metadata/MetadataUpdater.java +++ b/fluss-client/src/main/java/com/alibaba/fluss/client/metadata/MetadataUpdater.java @@ -51,6 +51,7 @@ import java.util.List; import java.util.Optional; import java.util.Set; +import java.util.concurrent.TimeoutException; import java.util.stream.Collectors; import static com.alibaba.fluss.client.utils.MetadataUtils.sendMetadataRequestAndRebuildCluster; @@ -59,6 +60,8 @@ public class MetadataUpdater { private static final Logger LOG = LoggerFactory.getLogger(MetadataUpdater.class); + private static final int MAX_RETRY_TIMES = 5; + private final RpcClient rpcClient; protected volatile Cluster cluster; @@ -119,8 +122,24 @@ public TableDescriptor getTableDescriptorOrElseThrow(long tableId) { public int leaderFor(TableBucket tableBucket) { ServerNode serverNode = cluster.leaderFor(tableBucket); if (serverNode == null) { - throw new FlussRuntimeException("Leader not found for table bucket: " + tableBucket); + for (int i = 0; i < MAX_RETRY_TIMES; i++) { + TablePath tablePath = cluster.getTablePathOrElseThrow(tableBucket.getTableId()); + updateMetadata(Collections.singleton(tablePath), null, null); + serverNode = cluster.leaderFor(tableBucket); + if (serverNode != null) { + break; + } + } + + if (serverNode == null) { + throw new FlussRuntimeException( + "Leader not found after retry " + + MAX_RETRY_TIMES + + " times for table bucket: " + + tableBucket); + } } + return serverNode.id(); } @@ -239,7 +258,7 @@ private void updateMetadata( } } catch (Exception e) { Throwable t = ExceptionUtils.stripExecutionException(e); - if (t instanceof RetriableException) { + if (t instanceof RetriableException || t instanceof TimeoutException) { LOG.warn("Failed to update metadata, but the exception is re-triable.", t); } else { throw new FlussRuntimeException("Failed to update metadata", t); diff --git a/fluss-client/src/main/java/com/alibaba/fluss/client/scanner/log/LogFetcher.java b/fluss-client/src/main/java/com/alibaba/fluss/client/scanner/log/LogFetcher.java index 0371c09e11..665ff091c2 100644 --- a/fluss-client/src/main/java/com/alibaba/fluss/client/scanner/log/LogFetcher.java +++ b/fluss-client/src/main/java/com/alibaba/fluss/client/scanner/log/LogFetcher.java @@ -91,6 +91,8 @@ public class LogFetcher implements Closeable { private final RpcClient rpcClient; private final int maxFetchBytes; private final int maxBucketFetchBytes; + private final int minFetchBytes; + private final int maxFetchWaitMs; private final boolean isCheckCrcs; private final LogScannerStatus logScannerStatus; private final LogFetchBuffer logFetchBuffer; @@ -123,9 +125,17 @@ public LogFetcher( this.projection = projection; this.rpcClient = rpcClient; this.logScannerStatus = logScannerStatus; - this.maxFetchBytes = (int) conf.get(ConfigOptions.LOG_FETCH_MAX_BYTES).getBytes(); + this.maxFetchBytes = + (int) conf.get(ConfigOptions.CLIENT_SCANNER_LOG_FETCH_MAX_BYTES).getBytes(); this.maxBucketFetchBytes = - (int) conf.get(ConfigOptions.LOG_FETCH_MAX_BYTES_FOR_BUCKET).getBytes(); + (int) + conf.get(ConfigOptions.CLIENT_SCANNER_LOG_FETCH_MAX_BYTES_FOR_BUCKET) + .getBytes(); + this.minFetchBytes = + (int) conf.get(ConfigOptions.CLIENT_SCANNER_LOG_FETCH_MIN_BYTES).getBytes(); + this.maxFetchWaitMs = + (int) conf.get(ConfigOptions.CLIENT_SCANNER_LOG_FETCH_WAIT_MAX_TIME).toMillis(); + this.isCheckCrcs = conf.getBoolean(ConfigOptions.CLIENT_SCANNER_LOG_CHECK_CRC); this.logFetchBuffer = new LogFetchBuffer(); this.nodesWithPendingFetchRequests = new HashSet<>(); @@ -419,7 +429,9 @@ private Map prepareFetchLogRequests() { FetchLogRequest fetchLogRequest = new FetchLogRequest() .setFollowerServerId(-1) - .setMaxBytes(maxFetchBytes); + .setMaxBytes(maxFetchBytes) + .setMinBytes(minFetchBytes) + .setMaxWaitMs(maxFetchWaitMs); PbFetchLogReqForTable reqForTable = new PbFetchLogReqForTable().setTableId(finalTableId); if (readContext.isProjectionPushDowned()) { diff --git a/fluss-client/src/main/java/com/alibaba/fluss/client/utils/MetadataUtils.java b/fluss-client/src/main/java/com/alibaba/fluss/client/utils/MetadataUtils.java index 8a51faf9b7..b4d011980f 100644 --- a/fluss-client/src/main/java/com/alibaba/fluss/client/utils/MetadataUtils.java +++ b/fluss-client/src/main/java/com/alibaba/fluss/client/utils/MetadataUtils.java @@ -47,6 +47,8 @@ import java.util.Random; import java.util.Set; import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; /** Utils for metadata for client. */ public class MetadataUtils { @@ -60,7 +62,7 @@ public class MetadataUtils { */ public static Cluster sendMetadataRequestAndRebuildCluster( AdminReadOnlyGateway gateway, Set tablePaths) - throws ExecutionException, InterruptedException { + throws ExecutionException, InterruptedException, TimeoutException { return sendMetadataRequestAndRebuildCluster(gateway, false, null, tablePaths, null, null); } @@ -76,7 +78,7 @@ public static Cluster sendMetadataRequestAndRebuildCluster( @Nullable Set tablePaths, @Nullable Collection tablePartitionNames, @Nullable Collection tablePartitionIds) - throws ExecutionException, InterruptedException { + throws ExecutionException, InterruptedException, TimeoutException { AdminReadOnlyGateway gateway = GatewayClientProxy.createGatewayProxy( () -> getOneAvailableTabletServerNode(cluster), @@ -94,7 +96,7 @@ public static Cluster sendMetadataRequestAndRebuildCluster( @Nullable Set tablePaths, @Nullable Collection tablePartitions, @Nullable Collection tablePartitionIds) - throws ExecutionException, InterruptedException { + throws ExecutionException, InterruptedException, TimeoutException { MetadataRequest metadataRequest = ClientRpcMessageUtils.makeMetadataRequest( tablePaths, tablePartitions, tablePartitionIds); @@ -151,7 +153,9 @@ public static Cluster sendMetadataRequestAndRebuildCluster( newPartitionIdByPath, newTablePathToTableInfo); }) - .get(); + .get(5, TimeUnit.SECONDS); // TODO currently, we don't have timeout logic in + // RpcClient, it will let the get() block forever. So we + // time out here } private static NewTableMetadata getTableMetadataToUpdate( diff --git a/fluss-common/src/main/java/com/alibaba/fluss/config/ConfigOptions.java b/fluss-common/src/main/java/com/alibaba/fluss/config/ConfigOptions.java index a7c8079c90..2c2893cdb7 100644 --- a/fluss-common/src/main/java/com/alibaba/fluss/config/ConfigOptions.java +++ b/fluss-common/src/main/java/com/alibaba/fluss/config/ConfigOptions.java @@ -385,6 +385,14 @@ public class ConfigOptions { "The purge number (in number of requests) of the write operation manager, " + "the default value is 1000."); + public static final ConfigOption LOG_REPLICA_FETCH_OPERATION_PURGE_NUMBER = + key("log.replica.fetch-operation-purge-number") + .intType() + .defaultValue(1000) + .withDescription( + "The purge number (in number of requests) of the fetch log operation manager, " + + "the default value is 1000."); + public static final ConfigOption LOG_REPLICA_FETCHER_NUMBER = key("log.replica.fetcher-number") .intType() @@ -397,31 +405,54 @@ public class ConfigOptions { + "tablet server at the cost of higher CPU and memory utilization."); public static final ConfigOption LOG_REPLICA_FETCH_BACKOFF_INTERVAL = - key("log.replica.fetch-backoff-interval") + key("log.replica.fetch.backoff-interval") .durationType() .defaultValue(Duration.ofSeconds(1)) - .withDescription("The amount of time to sleep when fetch bucket error occurs."); + .withDescription("The amount of time to sleep when fetch bucket error occurs.") + .withFallbackKeys("log.replica.fetch-backoff-interval"); - public static final ConfigOption LOG_FETCH_MAX_BYTES = - key("log.fetch.max-bytes") + public static final ConfigOption LOG_REPLICA_FETCH_MAX_BYTES = + key("log.replica.fetch.max-bytes") .memoryType() .defaultValue(MemorySize.parse("16mb")) .withDescription( - "The maximum amount of data the server should return for a fetch request. " - + "Records are fetched in batches for log scanner or follower, for one request batch, " - + "and if the first record batch in the first non-empty bucket of the fetch is " - + "larger than this value, the record batch will still be returned to ensure that " - + "the fetch can make progress. As such, this is not a absolute maximum. Note that " - + "the fetcher performs multiple fetches in parallel."); + "The maximum amount of data the server should return for a fetch request from follower. " + + "Records are fetched in batches, and if the first record batch in the first " + + "non-empty bucket of the fetch is larger than this value, the record batch " + + "will still be returned to ensure that the fetch can make progress. As such, " + + "this is not a absolute maximum. Note that the fetcher performs multiple fetches " + + "in parallel.") + .withDeprecatedKeys("log.fetch.max-bytes"); - public static final ConfigOption LOG_FETCH_MAX_BYTES_FOR_BUCKET = - key("log.fetch.max-bytes-for-bucket") + public static final ConfigOption LOG_REPLICA_FETCH_MAX_BYTES_FOR_BUCKET = + key("log.replica.fetch.max-bytes-for-bucket") .memoryType() .defaultValue(MemorySize.parse("1mb")) .withDescription( - "The maximum amount of data the server should return for a table bucket in fetch request. " - + "Records are fetched in batches for consumer or follower, for one request batch, " - + "the max bytes size is config by this option."); + "The maximum amount of data the server should return for a table bucket in fetch request " + + "from follower. Records are fetched in batches, the max bytes size is " + + "config by this option.") + .withDeprecatedKeys("log.fetch.max-bytes-for-bucket"); + + public static final ConfigOption LOG_REPLICA_FETCH_WAIT_MAX_TIME = + key("log.replica.fetch.wait-max-time") + .durationType() + .defaultValue(Duration.ofMillis(500)) + .withDescription( + "The maximum time to wait for enough bytes to be available for a fetch log request " + + "from follower to response. This value should always be less than the " + + "'log.replica.max-lag-time' at all times to prevent frequent shrinking of ISR for " + + "low throughput tables"); + + public static final ConfigOption LOG_REPLICA_FETCH_MIN_BYTES = + key("log.replica.fetch.min-bytes") + .memoryType() + .defaultValue(MemorySize.parse("1b")) + .withDescription( + "The minimum bytes expected for each fetch log request from follower to response. " + + "If not enough bytes, wait up to " + + LOG_REPLICA_FETCH_WAIT_MAX_TIME.key() + + " time to return."); public static final ConfigOption LOG_REPLICA_MIN_IN_SYNC_REPLICAS_NUMBER = key("log.replica.min-in-sync-replicas-number") @@ -714,6 +745,44 @@ public class ConfigOptions { + "The Scanner will cache the records from each fetch request and returns " + "them incrementally from each poll."); + public static final ConfigOption CLIENT_SCANNER_LOG_FETCH_MAX_BYTES = + key("client.scanner.log.fetch.max-bytes") + .memoryType() + .defaultValue(MemorySize.parse("16mb")) + .withDescription( + "The maximum amount of data the server should return for a fetch request from client. " + + "Records are fetched in batches, and if the first record batch in the first " + + "non-empty bucket of the fetch is larger than this value, the record batch " + + "will still be returned to ensure that the fetch can make progress. As such, " + + "this is not a absolute maximum."); + + public static final ConfigOption CLIENT_SCANNER_LOG_FETCH_MAX_BYTES_FOR_BUCKET = + key("client.scanner.log.fetch.max-bytes-for-bucket") + .memoryType() + .defaultValue(MemorySize.parse("1mb")) + .withDescription( + "The maximum amount of data the server should return for a table bucket in fetch request " + + "from client. Records are fetched in batches, the max bytes size is config by " + + "this option."); + + public static final ConfigOption CLIENT_SCANNER_LOG_FETCH_WAIT_MAX_TIME = + key("client.scanner.log.fetch.wait-max-time") + .durationType() + .defaultValue(Duration.ofMillis(500)) + .withDescription( + "The maximum time to wait for enough bytes to be available for a fetch log " + + "request from client to response."); + + public static final ConfigOption CLIENT_SCANNER_LOG_FETCH_MIN_BYTES = + key("client.scanner.log.fetch.min-bytes") + .memoryType() + .defaultValue(MemorySize.parse("1b")) + .withDescription( + "The minimum bytes expected for each fetch log request from client to response. " + + "If not enough bytes, wait up to " + + CLIENT_SCANNER_LOG_FETCH_WAIT_MAX_TIME.key() + + " time to return."); + public static final ConfigOption CLIENT_LOOKUP_QUEUE_SIZE = key("client.lookup.queue-size") .intType() diff --git a/fluss-common/src/main/java/com/alibaba/fluss/metrics/MetricNames.java b/fluss-common/src/main/java/com/alibaba/fluss/metrics/MetricNames.java index bfd1c92b1d..c94ab2f20a 100644 --- a/fluss-common/src/main/java/com/alibaba/fluss/metrics/MetricNames.java +++ b/fluss-common/src/main/java/com/alibaba/fluss/metrics/MetricNames.java @@ -48,7 +48,13 @@ public class MetricNames { public static final String REPLICA_LEADER_COUNT = "leaderCount"; public static final String REPLICA_COUNT = "replicaCount"; public static final String WRITE_ID_COUNT = "writerIdCount"; - public static final String DELAYED_OPERATIONS_SIZE = "delayedOperationsSize"; + public static final String DELAYED_WRITE_COUNT = "delayedWriteCount"; + public static final String DELAYED_WRITE_EXPIRES_RATE = "delayedWriteExpiresPerSecond"; + public static final String DELAYED_FETCH_COUNT = "delayedFetchCount"; + public static final String DELAYED_FETCH_FROM_FOLLOWER_EXPIRES_RATE = + "delayedFetchFromFollowerExpiresPerSecond"; + public static final String DELAYED_FETCH_FROM_CLIENT_EXPIRES_RATE = + "delayedFetchFromClientExpiresPerSecond"; // -------------------------------------------------------------------------------------------- // metrics for table diff --git a/fluss-rpc/src/main/proto/FlussApi.proto b/fluss-rpc/src/main/proto/FlussApi.proto index dea963661a..943e511fae 100644 --- a/fluss-rpc/src/main/proto/FlussApi.proto +++ b/fluss-rpc/src/main/proto/FlussApi.proto @@ -171,6 +171,8 @@ message FetchLogRequest { required int32 follower_server_id = 1; // value -1 indicate the request from client. required int32 max_bytes = 2; repeated PbFetchLogReqForTable tables_req = 3; + optional int32 max_wait_ms = 4; + optional int32 min_bytes = 5; } message FetchLogResponse { diff --git a/fluss-server/src/main/java/com/alibaba/fluss/server/log/FetchParams.java b/fluss-server/src/main/java/com/alibaba/fluss/server/log/FetchParams.java index bd768c0412..da57cf4528 100644 --- a/fluss-server/src/main/java/com/alibaba/fluss/server/log/FetchParams.java +++ b/fluss-server/src/main/java/com/alibaba/fluss/server/log/FetchParams.java @@ -18,6 +18,7 @@ import com.alibaba.fluss.annotation.VisibleForTesting; import com.alibaba.fluss.record.FileLogProjection; +import com.alibaba.fluss.rpc.messages.FetchLogRequest; import com.alibaba.fluss.types.RowType; import javax.annotation.Nullable; @@ -29,6 +30,21 @@ public final class FetchParams { /** Value -2L means we will fetch from log start offset. */ public static final long FETCH_FROM_EARLIEST_OFFSET = -2L; + /** + * Default min fetch bytes, which means the fetch request will be satisfied even if no bytes + * fetched. + */ + public static final int DEFAULT_MIN_FETCH_BYTES = -1; + + /** Default max wait ms, which means the fetch request will be satisfied immediately. */ + public static final long DEFAULT_MAX_WAIT_MS = -1L; + + /** + * Default max wait ms when log fetch minBytes set in {@link FetchLogRequest} but maxWaitMs not + * set. + */ + public static final long DEFAULT_MAX_WAIT_MS_WHEN_MIN_BYTES_ENABLE = 100L; + private final int replicaId; // Currently, FetchOnlyLeader can be set to false only for test, // which indicate that the client can read log data from follower. @@ -46,20 +62,33 @@ public final class FetchParams { // the lazily initialized projection util to read and project file logs @Nullable private FileLogProjection fileLogProjection; - // TODO: add more params like epoch, minBytes etc. + private final int minFetchBytes; + private final long maxWaitMs; + // TODO: add more params like epoch etc. public FetchParams(int replicaId, int maxFetchBytes) { - this(replicaId, true, maxFetchBytes); + this(replicaId, true, maxFetchBytes, DEFAULT_MIN_FETCH_BYTES, DEFAULT_MAX_WAIT_MS); + } + + public FetchParams(int replicaId, int maxFetchBytes, int minFetchBytes, long maxWaitMs) { + this(replicaId, true, maxFetchBytes, minFetchBytes, maxWaitMs); } @VisibleForTesting - public FetchParams(int replicaId, boolean fetchOnlyLeader, int maxFetchBytes) { + public FetchParams( + int replicaId, + boolean fetchOnlyLeader, + int maxFetchBytes, + int minFetchBytes, + long maxWaitMs) { this.replicaId = replicaId; this.fetchOnlyLeader = fetchOnlyLeader; this.maxFetchBytes = maxFetchBytes; this.fetchIsolation = FetchIsolation.of(replicaId >= 0); this.minOneMessage = true; this.fetchOffset = -1; + this.minFetchBytes = minFetchBytes; + this.maxWaitMs = maxWaitMs; } public void setCurrentFetch( @@ -118,6 +147,14 @@ public int maxFetchBytes() { return maxFetchBytes; } + public int minFetchBytes() { + return minFetchBytes; + } + + public long maxWaitMs() { + return maxWaitMs; + } + public int replicaId() { return replicaId; } @@ -143,12 +180,15 @@ public boolean equals(Object o) { return false; } FetchParams that = (FetchParams) o; - return replicaId == that.replicaId && maxFetchBytes == that.maxFetchBytes; + return replicaId == that.replicaId + && maxFetchBytes == that.maxFetchBytes + && minFetchBytes == that.minFetchBytes + && maxWaitMs == that.maxWaitMs; } @Override public int hashCode() { - return Objects.hash(replicaId, maxFetchBytes); + return Objects.hash(replicaId, maxFetchBytes, minFetchBytes, maxWaitMs); } @Override @@ -158,6 +198,10 @@ public String toString() { + replicaId + ", maxFetchBytes=" + maxFetchBytes + + ", minFetchBytes=" + + minFetchBytes + + ", maxWaitMs=" + + maxWaitMs + ')'; } } diff --git a/fluss-server/src/main/java/com/alibaba/fluss/server/log/LogOffsetMetadata.java b/fluss-server/src/main/java/com/alibaba/fluss/server/log/LogOffsetMetadata.java index 956021f9c9..55377e54b0 100644 --- a/fluss-server/src/main/java/com/alibaba/fluss/server/log/LogOffsetMetadata.java +++ b/fluss-server/src/main/java/com/alibaba/fluss/server/log/LogOffsetMetadata.java @@ -67,6 +67,30 @@ public int getRelativePositionInSegment() { return relativePositionInSegment; } + /** + * Compute the number of bytes between this offset to the given offset, if they are on the same + * segment and this offset precedes the given offset. + */ + public int positionDiff(LogOffsetMetadata that) { + if (messageOffsetOnly()) { + throw new FlussRuntimeException( + this + + " cannot compare its message offset with " + + that + + " since it only has message offset info."); + } + + if (!onSameSegment(that)) { + throw new FlussRuntimeException( + this + + " is not on the same segment with " + + that + + " since they are not on the same segment."); + } + + return this.relativePositionInSegment - that.relativePositionInSegment; + } + // check if this offset is already on an older segment compared with the given offset. public boolean onOlderSegment(LogOffsetMetadata that) { if (messageOffsetOnly()) { @@ -85,6 +109,11 @@ public boolean messageOffsetOnly() { && relativePositionInSegment == UNKNOWN_FILE_POSITION; } + // check if this offset is on the same segment with the given offset + private boolean onSameSegment(LogOffsetMetadata that) { + return this.segmentBaseOffset == that.segmentBaseOffset; + } + @Override public String toString() { return "(offset=" diff --git a/fluss-server/src/main/java/com/alibaba/fluss/server/log/LogOffsetSnapshot.java b/fluss-server/src/main/java/com/alibaba/fluss/server/log/LogOffsetSnapshot.java new file mode 100644 index 0000000000..0107077249 --- /dev/null +++ b/fluss-server/src/main/java/com/alibaba/fluss/server/log/LogOffsetSnapshot.java @@ -0,0 +1,79 @@ +/* + * Copyright (c) 2024 Alibaba Group Holding Ltd. + * + * Licensed 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 com.alibaba.fluss.server.log; + +/* This file is based on source code of Apache Kafka Project (https://kafka.apache.org/), licensed by the Apache + * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for + * additional information regarding copyright ownership. */ + +/** A snapshot of the different log offsets. */ +public class LogOffsetSnapshot { + public final long logStartOffset; + public final long localLogStartOffset; + public final LogOffsetMetadata logEndOffset; + public final LogOffsetMetadata highWatermark; + + public LogOffsetSnapshot( + long logStartOffset, + long localLogStartOffset, + LogOffsetMetadata logEndOffset, + LogOffsetMetadata highWatermark) { + this.logStartOffset = logStartOffset; + this.localLogStartOffset = localLogStartOffset; + this.logEndOffset = logEndOffset; + this.highWatermark = highWatermark; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + + LogOffsetSnapshot that = (LogOffsetSnapshot) o; + return logStartOffset == that.logStartOffset + && localLogStartOffset == that.localLogStartOffset + && logEndOffset.equals(that.logEndOffset) + && highWatermark.equals(that.highWatermark); + } + + @Override + public int hashCode() { + int result = Long.hashCode(logStartOffset); + result = 31 * result + Long.hashCode(localLogStartOffset); + result = 31 * result + logEndOffset.hashCode(); + result = 31 * result + highWatermark.hashCode(); + return result; + } + + @Override + public String toString() { + return "LogOffsetSnapshot(" + + "logStartOffset=" + + logStartOffset + + ", localLogStartOffset=" + + localLogStartOffset + + ", logEndOffset=" + + logEndOffset + + ", highWatermark=" + + highWatermark + + ')'; + } +} diff --git a/fluss-server/src/main/java/com/alibaba/fluss/server/log/LogTablet.java b/fluss-server/src/main/java/com/alibaba/fluss/server/log/LogTablet.java index 4e40a6e176..dbae560fe7 100644 --- a/fluss-server/src/main/java/com/alibaba/fluss/server/log/LogTablet.java +++ b/fluss-server/src/main/java/com/alibaba/fluss/server/log/LogTablet.java @@ -500,6 +500,20 @@ public void deleteSegmentsAlreadyExistsInRemote() { deleteSegments(remoteLogEndOffset); } + /** + * Fully materialize and return an offset snapshot including segment position info. This method + * will update the LogOffsetMetadata for the high watermark if they are message-only. Throws an + * offset out of range error if the segment info cannot be loaded. + */ + public LogOffsetSnapshot fetchOffsetSnapshot() throws IOException { + LogOffsetMetadata highWatermark = fetchHighWatermarkMetadata(); + return new LogOffsetSnapshot( + logStartOffset(), + localLogStartOffset(), + localLog.getLocalLogEndOffsetMetadata(), + highWatermark); + } + private void deleteSegments(long cleanUpToOffset) { // cache to local variables long localLogStartOffset = localLog.getLocalLogStartOffset(); diff --git a/fluss-server/src/main/java/com/alibaba/fluss/server/metrics/group/TabletServerMetricGroup.java b/fluss-server/src/main/java/com/alibaba/fluss/server/metrics/group/TabletServerMetricGroup.java index 2633204701..eb65207949 100644 --- a/fluss-server/src/main/java/com/alibaba/fluss/server/metrics/group/TabletServerMetricGroup.java +++ b/fluss-server/src/main/java/com/alibaba/fluss/server/metrics/group/TabletServerMetricGroup.java @@ -43,6 +43,9 @@ public class TabletServerMetricGroup extends AbstractMetricGroup { // ---- metrics ---- private final Counter replicationBytesIn; private final Counter replicationBytesOut; + private final Counter delayedWriteExpireCount; + private final Counter delayedFetchFromFollowerExpireCount; + private final Counter delayedFetchFromClientExpireCount; public TabletServerMetricGroup( MetricRegistry registry, String clusterId, String hostname, int serverId) { @@ -55,6 +58,17 @@ public TabletServerMetricGroup( meter(MetricNames.REPLICATION_IN_RATE, new MeterView(replicationBytesIn)); replicationBytesOut = new ThreadSafeSimpleCounter(); meter(MetricNames.REPLICATION_OUT_RATE, new MeterView(replicationBytesOut)); + + delayedWriteExpireCount = new ThreadSafeSimpleCounter(); + meter(MetricNames.DELAYED_WRITE_EXPIRES_RATE, new MeterView(delayedWriteExpireCount)); + delayedFetchFromFollowerExpireCount = new ThreadSafeSimpleCounter(); + meter( + MetricNames.DELAYED_FETCH_FROM_FOLLOWER_EXPIRES_RATE, + new MeterView(delayedFetchFromFollowerExpireCount)); + delayedFetchFromClientExpireCount = new ThreadSafeSimpleCounter(); + meter( + MetricNames.DELAYED_FETCH_FROM_CLIENT_EXPIRES_RATE, + new MeterView(delayedFetchFromClientExpireCount)); } @Override @@ -77,6 +91,18 @@ public Counter replicationBytesOut() { return replicationBytesOut; } + public Counter delayedWriteExpireCount() { + return delayedWriteExpireCount; + } + + public Counter delayedFetchFromFollowerExpireCount() { + return delayedFetchFromFollowerExpireCount; + } + + public Counter delayedFetchFromClientExpireCount() { + return delayedFetchFromClientExpireCount; + } + // ------------------------------------------------------------------------ // table buckets groups // ------------------------------------------------------------------------ diff --git a/fluss-server/src/main/java/com/alibaba/fluss/server/replica/Replica.java b/fluss-server/src/main/java/com/alibaba/fluss/server/replica/Replica.java index 77dcba13a5..6d3c80e726 100644 --- a/fluss-server/src/main/java/com/alibaba/fluss/server/replica/Replica.java +++ b/fluss-server/src/main/java/com/alibaba/fluss/server/replica/Replica.java @@ -67,6 +67,7 @@ import com.alibaba.fluss.server.log.LogAppendInfo; import com.alibaba.fluss.server.log.LogManager; import com.alibaba.fluss.server.log.LogOffsetMetadata; +import com.alibaba.fluss.server.log.LogOffsetSnapshot; import com.alibaba.fluss.server.log.LogReadInfo; import com.alibaba.fluss.server.log.LogTablet; import com.alibaba.fluss.server.log.checkpoint.OffsetCheckpointFile; @@ -74,9 +75,10 @@ import com.alibaba.fluss.server.metadata.ServerMetadataCache; import com.alibaba.fluss.server.metrics.group.BucketMetricGroup; import com.alibaba.fluss.server.metrics.group.PhysicalTableMetricGroup; +import com.alibaba.fluss.server.replica.delay.DelayedFetchLog; import com.alibaba.fluss.server.replica.delay.DelayedOperationManager; +import com.alibaba.fluss.server.replica.delay.DelayedTableBucketKey; import com.alibaba.fluss.server.replica.delay.DelayedWrite; -import com.alibaba.fluss.server.replica.delay.DelayedWriteKey; import com.alibaba.fluss.server.utils.FatalErrorHandler; import com.alibaba.fluss.server.zk.ZkSequenceIDCounter; import com.alibaba.fluss.server.zk.ZooKeeperClient; @@ -157,6 +159,7 @@ public final class Replica { private final int localTabletServerId; private final DelayedOperationManager> delayedWriteManager; + private final DelayedOperationManager delayedFetchLogManager; /** The manger to manger the isr expand and shrink. */ private final AdjustIsrManager adjustIsrManager; @@ -202,6 +205,7 @@ public Replica( int localTabletServerId, OffsetCheckpointFile.LazyOffsetCheckpoints lazyHighWatermarkCheckpoint, DelayedOperationManager> delayedWriteManager, + DelayedOperationManager delayedFetchLogManager, AdjustIsrManager adjustIsrManager, SnapshotContext snapshotContext, ServerMetadataCache metadataCache, @@ -218,6 +222,7 @@ public Replica( this.minInSyncReplicas = minInSyncReplicas; this.localTabletServerId = localTabletServerId; this.delayedWriteManager = delayedWriteManager; + this.delayedFetchLogManager = delayedFetchLogManager; this.adjustIsrManager = adjustIsrManager; this.fatalErrorHandler = fatalErrorHandler; this.bucketMetricGroup = bucketMetricGroup; @@ -483,6 +488,15 @@ public void checkProjection(@Nullable int[] projectedFields) { } } + public LogOffsetSnapshot fetchOffsetSnapshot(boolean fetchOnlyFromLeader) throws IOException { + return inReadLock( + leaderIsrUpdateLock, + () -> { + LogTablet logTablet = localLogOrThrow(fetchOnlyFromLeader); + return logTablet.fetchOffsetSnapshot(); + }); + } + // ------------------------------------------------------------------------------------------- private void onBecomeNewLeader() { @@ -779,8 +793,14 @@ public LogAppendInfo appendRecordsToLeader(MemoryLogRecords memoryLogRecords, in // TODO WRITE a leader epoch. LogAppendInfo appendInfo = logTablet.appendAsLeader(memoryLogRecords); - // we may need to increment high watermark. - maybeIncrementLeaderHW(logTablet, System.currentTimeMillis()); + // we may need to increment high watermark if isr could be down to 1 or the + // replica count is 1. + boolean hwIncreased = + maybeIncrementLeaderHW(logTablet, System.currentTimeMillis()); + + if (hwIncreased) { + tryCompleteDelayedOperations(); + } return appendInfo; }); @@ -1264,8 +1284,9 @@ private LogReadInfo readRecords(FetchParams fetchParams, LogTablet logTablet) } private void tryCompleteDelayedOperations() { - DelayedWriteKey delayedWriteKey = new DelayedWriteKey(tableBucket); - delayedWriteManager.checkAndComplete(delayedWriteKey); + DelayedTableBucketKey delayedTableBucketKey = new DelayedTableBucketKey(tableBucket); + delayedWriteManager.checkAndComplete(delayedTableBucketKey); + delayedFetchLogManager.checkAndComplete(delayedTableBucketKey); } private void validateBucketEpoch(int requestBucketEpoch) { diff --git a/fluss-server/src/main/java/com/alibaba/fluss/server/replica/ReplicaManager.java b/fluss-server/src/main/java/com/alibaba/fluss/server/replica/ReplicaManager.java index 957f549c02..c3fbd3e91f 100644 --- a/fluss-server/src/main/java/com/alibaba/fluss/server/replica/ReplicaManager.java +++ b/fluss-server/src/main/java/com/alibaba/fluss/server/replica/ReplicaManager.java @@ -71,10 +71,12 @@ import com.alibaba.fluss.server.kv.snapshot.CompletedKvSnapshotCommitter; import com.alibaba.fluss.server.kv.snapshot.DefaultSnapshotContext; import com.alibaba.fluss.server.kv.snapshot.SnapshotContext; +import com.alibaba.fluss.server.log.FetchDataInfo; import com.alibaba.fluss.server.log.FetchParams; import com.alibaba.fluss.server.log.ListOffsetsParam; import com.alibaba.fluss.server.log.LogAppendInfo; import com.alibaba.fluss.server.log.LogManager; +import com.alibaba.fluss.server.log.LogOffsetMetadata; import com.alibaba.fluss.server.log.LogReadInfo; import com.alibaba.fluss.server.log.LogTablet; import com.alibaba.fluss.server.log.checkpoint.OffsetCheckpointFile; @@ -83,9 +85,11 @@ import com.alibaba.fluss.server.metrics.group.BucketMetricGroup; import com.alibaba.fluss.server.metrics.group.PhysicalTableMetricGroup; import com.alibaba.fluss.server.metrics.group.TabletServerMetricGroup; +import com.alibaba.fluss.server.replica.delay.DelayedFetchLog; +import com.alibaba.fluss.server.replica.delay.DelayedFetchLog.FetchBucketStatus; import com.alibaba.fluss.server.replica.delay.DelayedOperationManager; +import com.alibaba.fluss.server.replica.delay.DelayedTableBucketKey; import com.alibaba.fluss.server.replica.delay.DelayedWrite; -import com.alibaba.fluss.server.replica.delay.DelayedWriteKey; import com.alibaba.fluss.server.replica.fetcher.InitialFetchStatus; import com.alibaba.fluss.server.replica.fetcher.ReplicaFetcherManager; import com.alibaba.fluss.server.utils.FatalErrorHandler; @@ -144,9 +148,20 @@ public class ReplicaManager { private final ServerMetadataCache metadataCache; private final Lock replicaStateChangeLock = new ReentrantLock(); - // delayed write operation manager is used to manage the delayed write operation, which is - // waited for other follower replicas ack. + + /** + * delayed write operation manager is used to manage the delayed write operation, which is + * waited for other follower replicas ack. + */ private final DelayedOperationManager> delayedWriteManager; + + /** + * delayed fetch log operation manager is used to manage the delayed fetch log operation, which + * is waited for the available fetch log size bigger than the minLogFetchSize or the wait time + * is up. + */ + private final DelayedOperationManager delayedFetchLogManager; + private final ReplicaFetcherManager replicaFetcherManager; // The manager used to manager the replica alter, especially the isr expand and shrink. private final AdjustIsrManager adjustIsrManager; @@ -229,8 +244,13 @@ public ReplicaManager( new DelayedOperationManager<>( "delay write", serverId, - conf.getInt(ConfigOptions.LOG_REPLICA_WRITE_OPERATION_PURGE_NUMBER), - serverMetricGroup); + conf.getInt(ConfigOptions.LOG_REPLICA_WRITE_OPERATION_PURGE_NUMBER)); + this.delayedFetchLogManager = + new DelayedOperationManager<>( + "delay fetch log", + serverId, + conf.getInt(ConfigOptions.LOG_REPLICA_FETCH_OPERATION_PURGE_NUMBER)); + this.replicaFetcherManager = new ReplicaFetcherManager(conf, rpcClient, serverId, this); this.adjustIsrManager = new AdjustIsrManager(scheduler, coordinatorGateway, serverId); this.fatalErrorHandler = fatalErrorHandler; @@ -266,6 +286,9 @@ private void registerMetrics() { () -> onlineReplicas().filter(Replica::isLeader).count()); serverMetricGroup.gauge(MetricNames.REPLICA_COUNT, allReplicas::size); serverMetricGroup.gauge(MetricNames.WRITE_ID_COUNT, this::writerIdCount); + serverMetricGroup.gauge(MetricNames.DELAYED_WRITE_COUNT, delayedWriteManager::numDelayed); + serverMetricGroup.gauge( + MetricNames.DELAYED_FETCH_COUNT, delayedFetchLogManager::numDelayed); } private Stream onlineReplicas() { @@ -370,14 +393,15 @@ public void fetchLogRecords( Map bucketFetchInfo, Consumer> responseCallback) { long startTime = System.currentTimeMillis(); - Map logFetchResults = read(params, bucketFetchInfo); + Map logReadResults = readFromLog(params, bucketFetchInfo); if (LOG.isTraceEnabled()) { LOG.trace( "Fetch log records from local log in {} ms", System.currentTimeMillis() - startTime); } - // TODO add delay fetch logic. - responseCallback.accept(logFetchResults); + + // maybe do delay fetch log operation. + maybeAddDelayedFetchLog(params, bucketFetchInfo, logReadResults, responseCallback); } /** @@ -757,7 +781,7 @@ private void makeFollowers( .collect(Collectors.toSet())); replicasBecomeFollower.forEach( - replica -> completeDelayedWriteOperations(replica.getTableBucket())); + replica -> completeDelayedOperations(replica.getTableBucket())); LOG.info( "Stopped fetchers as part of become follower request for {} replicas", @@ -926,9 +950,9 @@ public void limitScan( responseCallback.accept(limitScanResultForBucket); } - private Map read( + public Map readFromLog( FetchParams fetchParams, Map bucketFetchInfo) { - Map logFetchResult = new HashMap<>(); + Map logReadResult = new HashMap<>(); boolean isFromFollower = fetchParams.isFromFollower(); int limitBytes = fetchParams.maxFetchBytes(); for (Map.Entry entry : bucketFetchInfo.entrySet()) { @@ -957,18 +981,19 @@ private Map read( // Once we read from a non-empty bucket, we stop ignoring request and bucket // level size limits. - int recordBatchSize = readInfo.getFetchedData().getRecords().sizeInBytes(); + FetchDataInfo fetchedData = readInfo.getFetchedData(); + int recordBatchSize = fetchedData.getRecords().sizeInBytes(); if (recordBatchSize > 0) { fetchParams.markReadOneMessage(); } limitBytes = Math.max(0, limitBytes - recordBatchSize); - logFetchResult.put( + logReadResult.put( tb, - new FetchLogResultForBucket( - tb, - readInfo.getFetchedData().getRecords(), - readInfo.getHighWatermark())); + new LogReadResult( + new FetchLogResultForBucket( + tb, fetchedData.getRecords(), readInfo.getHighWatermark()), + fetchedData.getFetchOffsetMetadata())); // update metrics if (isFromFollower) { @@ -995,10 +1020,11 @@ private Map read( } else { result = new FetchLogResultForBucket(tb, ApiError.fromThrowable(e)); } - logFetchResult.put(tb, result); + logReadResult.put( + tb, new LogReadResult(result, LogOffsetMetadata.UNKNOWN_OFFSET_METADATA)); } } - return logFetchResult; + return logReadResult; } private FetchLogResultForBucket handleFetchOutOfRangeException( @@ -1145,7 +1171,8 @@ private void maybeAddDelayedWrite( timeoutMs, new DelayedWrite.DelayedWriteMetadata<>(requiredAcks, bucketStatusMap), this, - responseCallback); + responseCallback, + serverMetricGroup); // try to complete the request immediately, otherwise put it into the manager. // This is because while the delayed write operation is being created, new @@ -1153,16 +1180,83 @@ private void maybeAddDelayedWrite( delayedWriteManager.tryCompleteElseWatch( delayedWrite, bucketStatusMap.keySet().stream() - .map(DelayedWriteKey::new) + .map(DelayedTableBucketKey::new) .collect(Collectors.toList())); } else { responseCallback.accept(new ArrayList<>(writeResults.values())); } } - private void completeDelayedWriteOperations(TableBucket tableBucket) { - DelayedWriteKey delayedWriteKey = new DelayedWriteKey(tableBucket); - delayedWriteManager.checkAndComplete(delayedWriteKey); + private void maybeAddDelayedFetchLog( + FetchParams params, + Map bucketFetchInfo, + Map logReadResults, + Consumer> responseCallback) { + long bytesReadable = 0; + boolean errorReadingData = false; + boolean hasFetchFromLocal = false; + Map fetchBucketStatusMap = new HashMap<>(); + for (Map.Entry logReadResultEntry : logReadResults.entrySet()) { + TableBucket tb = logReadResultEntry.getKey(); + LogReadResult logReadResult = logReadResultEntry.getValue(); + FetchLogResultForBucket fetchLogResultForBucket = + logReadResult.getFetchLogResultForBucket(); + if (fetchLogResultForBucket.failed()) { + errorReadingData = true; + break; + } + + if (!fetchLogResultForBucket.fetchFromRemote()) { + hasFetchFromLocal = true; + bytesReadable += fetchLogResultForBucket.recordsOrEmpty().sizeInBytes(); + } + + fetchBucketStatusMap.put( + tb, + new FetchBucketStatus( + bucketFetchInfo.get(tb), + logReadResult.getLogOffsetMetadata(), + fetchLogResultForBucket)); + } + + if (!hasFetchFromLocal + || params.maxWaitMs() <= 0 + || bucketFetchInfo.isEmpty() + || bytesReadable >= params.minFetchBytes() + || errorReadingData) { + responseCallback.accept( + logReadResults.entrySet().stream() + .collect( + Collectors.toMap( + Map.Entry::getKey, + entry -> + entry.getValue() + .getFetchLogResultForBucket()))); + } else { + // need to put the fetch log request into delayed fetch log manager. + DelayedFetchLog delayedFetchLog = + new DelayedFetchLog( + params, + this, + fetchBucketStatusMap, + responseCallback, + serverMetricGroup); + + // try to complete the request immediately, otherwise put it into the + // delayedFetchLogManager; this is because while the delayed fetch log operation is + // being created, new requests may arrive and hence make this operation completable. + delayedFetchLogManager.tryCompleteElseWatch( + delayedFetchLog, + fetchBucketStatusMap.keySet().stream() + .map(DelayedTableBucketKey::new) + .collect(Collectors.toList())); + } + } + + private void completeDelayedOperations(TableBucket tableBucket) { + DelayedTableBucketKey delayedTableBucketKey = new DelayedTableBucketKey(tableBucket); + delayedWriteManager.checkAndComplete(delayedTableBucketKey); + delayedFetchLogManager.checkAndComplete(delayedTableBucketKey); } /** @@ -1279,7 +1373,7 @@ private StopReplicaResultForBucket stopReplica( // If we were the leader, we may have some operations still waiting for completion. // We force completion to prevent them from timing out. - completeDelayedWriteOperations(tb); + completeDelayedOperations(tb); return new StopReplicaResultForBucket(tb); } @@ -1346,6 +1440,7 @@ protected Optional maybeCreateReplica(NotifyLeaderAndIsrData data) { new OffsetCheckpointFile.LazyOffsetCheckpoints( highWatermarkCheckpoint), delayedWriteManager, + delayedFetchLogManager, adjustIsrManager, kvSnapshotContext, metadataCache, @@ -1417,6 +1512,11 @@ public DelayedOperationManager> getDelayedWriteManager() { return delayedWriteManager; } + @VisibleForTesting + public DelayedOperationManager getDelayedFetchLogManager() { + return delayedFetchLogManager; + } + @VisibleForTesting public AdjustIsrManager getAdjustIsrManager() { return adjustIsrManager; @@ -1457,8 +1557,30 @@ public void shutdown() throws InterruptedException { kvSnapshotResource.close(); replicaFetcherManager.shutdown(); delayedWriteManager.shutdown(); + delayedFetchLogManager.shutdown(); // Checkpoint highWatermark. checkpointHighWatermarks(); } + + /** The result of reading log. */ + public static final class LogReadResult { + private final FetchLogResultForBucket fetchLogResultForBucket; + private final LogOffsetMetadata logOffsetMetadata; + + public LogReadResult( + FetchLogResultForBucket fetchLogResultForBucket, + LogOffsetMetadata logOffsetMetadata) { + this.fetchLogResultForBucket = fetchLogResultForBucket; + this.logOffsetMetadata = logOffsetMetadata; + } + + public FetchLogResultForBucket getFetchLogResultForBucket() { + return fetchLogResultForBucket; + } + + public LogOffsetMetadata getLogOffsetMetadata() { + return logOffsetMetadata; + } + } } diff --git a/fluss-server/src/main/java/com/alibaba/fluss/server/replica/delay/DelayedFetchLog.java b/fluss-server/src/main/java/com/alibaba/fluss/server/replica/delay/DelayedFetchLog.java new file mode 100644 index 0000000000..80b145d5af --- /dev/null +++ b/fluss-server/src/main/java/com/alibaba/fluss/server/replica/delay/DelayedFetchLog.java @@ -0,0 +1,241 @@ +/* + * Copyright (c) 2024 Alibaba Group Holding Ltd. + * + * Licensed 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 com.alibaba.fluss.server.replica.delay; + +/* This file is based on source code of Apache Kafka Project (https://kafka.apache.org/), licensed by the Apache + * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for + * additional information regarding copyright ownership. */ + +import com.alibaba.fluss.exception.FlussRuntimeException; +import com.alibaba.fluss.exception.NotLeaderOrFollowerException; +import com.alibaba.fluss.exception.UnknownTableOrBucketException; +import com.alibaba.fluss.metadata.TableBucket; +import com.alibaba.fluss.rpc.entity.FetchLogResultForBucket; +import com.alibaba.fluss.rpc.messages.FetchLogRequest; +import com.alibaba.fluss.server.entity.FetchData; +import com.alibaba.fluss.server.log.FetchIsolation; +import com.alibaba.fluss.server.log.FetchParams; +import com.alibaba.fluss.server.log.LogOffsetMetadata; +import com.alibaba.fluss.server.log.LogOffsetSnapshot; +import com.alibaba.fluss.server.metrics.group.TabletServerMetricGroup; +import com.alibaba.fluss.server.replica.Replica; +import com.alibaba.fluss.server.replica.ReplicaManager; +import com.alibaba.fluss.server.replica.ReplicaManager.LogReadResult; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.HashMap; +import java.util.Map; +import java.util.function.Consumer; + +/** + * A delayed fetch log operation that can be created by the {@link ReplicaManager} and watched in + * the delayed fetch log operation manager. delayed fetch log operation can be sent by {@link + * FetchLogRequest}. + */ +public class DelayedFetchLog extends DelayedOperation { + + private static final Logger LOG = LoggerFactory.getLogger(DelayedFetchLog.class); + + private final FetchParams params; + private final ReplicaManager replicaManager; + private final Map fetchBucketStatusMap; + private final Consumer> responseCallback; + private final TabletServerMetricGroup serverMetricGroup; + + public DelayedFetchLog( + FetchParams params, + ReplicaManager replicaManager, + Map fetchBucketStatusMap, + Consumer> responseCallback, + TabletServerMetricGroup serverMetricGroup) { + super(params.maxWaitMs()); + this.params = params; + this.replicaManager = replicaManager; + this.fetchBucketStatusMap = fetchBucketStatusMap; + this.responseCallback = responseCallback; + this.serverMetricGroup = serverMetricGroup; + } + + /** Upon completion, read whatever data is available and pass to the complete callback. */ + @Override + public void onComplete() { + Map result = new HashMap<>(); + + Map reFetchBuckets = new HashMap<>(); + for (Map.Entry fetchBucketStatusEntry : + fetchBucketStatusMap.entrySet()) { + FetchBucketStatus fetchBucketStatus = fetchBucketStatusEntry.getValue(); + TableBucket tb = fetchBucketStatusEntry.getKey(); + if (fetchBucketStatus.previousFetchLogResultForBucket.fetchFromRemote()) { + result.put(tb, fetchBucketStatus.previousFetchLogResultForBucket); + } else { + reFetchBuckets.put(tb, fetchBucketStatus.fetchData); + } + } + + // re-fetch data. + Map reReadResult = + replicaManager.readFromLog(params, reFetchBuckets); + reReadResult.forEach((key, value) -> result.put(key, value.getFetchLogResultForBucket())); + responseCallback.accept(result); + } + + /** + * The delayed fetch operation can be completed if: + * + *
    + *
  • Case A: The server is no longer the leader for some buckets it tries to fetch + *
  • Case B: The replica is no longer available on this server + *
  • Case C: This server doesn't know of some buckets it tries to fetch + *
  • Case D: The fetch offset locates not on the last segment of the log + *
  • Case E: The accumulated bytes from all the fetching buckets exceeds the minimum bytes + *
+ * + *

Upon completion, should return whatever data is available for each valid bucket. + */ + @Override + public boolean tryComplete() { + int accumulatedSize = 0; + + for (Map.Entry entry : fetchBucketStatusMap.entrySet()) { + TableBucket tb = entry.getKey(); + FetchBucketStatus fetchBucketStatus = entry.getValue(); + LogOffsetMetadata fetchOffset = fetchBucketStatus.startOffsetMetadata; + try { + if (!fetchBucketStatus.previousFetchLogResultForBucket.fetchFromRemote() + && fetchOffset != LogOffsetMetadata.UNKNOWN_OFFSET_METADATA) { + Replica replica = replicaManager.getReplicaOrException(tb); + LogOffsetSnapshot logOffsetSnapshot = + replica.fetchOffsetSnapshot(params.fetchOnlyLeader()); + LogOffsetMetadata endOffset; + if (params.isolation() == FetchIsolation.LOG_END) { + endOffset = logOffsetSnapshot.logEndOffset; + } else if (params.isolation() == FetchIsolation.HIGH_WATERMARK) { + endOffset = logOffsetSnapshot.highWatermark; + } else { + throw new FlussRuntimeException("Unknown fetch isolation."); + } + + // Go directly to the check for Case E if the log offsets are the same. If + // the log segment has just rolled, then the high watermark offset will remain + // the same but be on the old segment, which would incorrectly be seen as an + // instance of Case D. + if (endOffset.getMessageOffset() != fetchOffset.getMessageOffset()) { + if (endOffset.onOlderSegment(fetchOffset)) { + // Case D, this can happen when the new fetch log operation is on a + // truncated leader. + LOG.debug( + "Satisfying delayed fetch log since it is fetching later segments of bucket {}.", + tb); + return forceComplete(); + } else if (fetchOffset.onOlderSegment(endOffset)) { + // Case D, this can happen when the fetch operation is falling behind + // the current segment or the bucket has just rolled a new segment. + LOG.debug( + "Satisfying delayed fetch log since it is fetching older segments of bucket {}.", + tb); + return forceComplete(); + } else if (fetchOffset.getMessageOffset() < endOffset.getMessageOffset()) { + // We take the bucket fetch size as upper bound when accumulating the + // bytes. + int bytesAvailable = + Math.min( + endOffset.positionDiff(fetchOffset), + fetchBucketStatus.fetchData.getMaxBytes()); + accumulatedSize += bytesAvailable; + } + } + } + } catch (NotLeaderOrFollowerException e) { + // case A and B. + LOG.debug( + "TabletServer is no longer the leader or follower of table-bucket {}, satisfy delayFetchLog immediately.", + tb); + return forceComplete(); + } catch (UnknownTableOrBucketException e) { + // case C + LOG.debug( + "TabletServer os mp longer knows of table-bucket {}, satisfy delayFetchLog immediately.", + tb); + return forceComplete(); + } catch (IOException e) { + LOG.debug( + "There is an storage exception append for table-bucket {}, satisfy delayFetchLog immediately.", + tb, + e); + return forceComplete(); + } + } + + // Case F. + if (accumulatedSize >= params.minFetchBytes()) { + return forceComplete(); + } else { + return false; + } + } + + @Override + public void onExpiration() { + if (params.isFromFollower()) { + serverMetricGroup.delayedFetchFromFollowerExpireCount().inc(); + } else { + serverMetricGroup.delayedFetchFromClientExpireCount().inc(); + } + } + + @Override + public String toString() { + return "DelayedFetchLog{" + + "params=" + + params + + ", numBuckets=" + + fetchBucketStatusMap.size() + + '}'; + } + + /** The status of a bucket in a delayed log fetch operation. */ + public static final class FetchBucketStatus { + private final FetchData fetchData; + private final LogOffsetMetadata startOffsetMetadata; + private final FetchLogResultForBucket previousFetchLogResultForBucket; + + public FetchBucketStatus( + FetchData fetchData, + LogOffsetMetadata startOffsetMetadata, + FetchLogResultForBucket previousFetchLogResultForBucket) { + this.fetchData = fetchData; + this.startOffsetMetadata = startOffsetMetadata; + this.previousFetchLogResultForBucket = previousFetchLogResultForBucket; + } + + @Override + public String toString() { + return "FetchBucketStatus{" + + "fetchData=" + + fetchData + + ", startOffsetMetadata=" + + startOffsetMetadata + + ", previousFetchLogResultForBucket=" + + previousFetchLogResultForBucket + + '}'; + } + } +} diff --git a/fluss-server/src/main/java/com/alibaba/fluss/server/replica/delay/DelayedOperationManager.java b/fluss-server/src/main/java/com/alibaba/fluss/server/replica/delay/DelayedOperationManager.java index 48f0467441..debfc24387 100644 --- a/fluss-server/src/main/java/com/alibaba/fluss/server/replica/delay/DelayedOperationManager.java +++ b/fluss-server/src/main/java/com/alibaba/fluss/server/replica/delay/DelayedOperationManager.java @@ -18,8 +18,6 @@ import com.alibaba.fluss.annotation.VisibleForTesting; import com.alibaba.fluss.exception.FlussRuntimeException; -import com.alibaba.fluss.metrics.MetricNames; -import com.alibaba.fluss.metrics.groups.MetricGroup; import com.alibaba.fluss.server.utils.timer.DefaultTimer; import com.alibaba.fluss.server.utils.timer.Timer; import com.alibaba.fluss.server.utils.timer.TimerTask; @@ -65,8 +63,7 @@ public final class DelayedOperationManager { private final AtomicInteger estimatedTotalOperations = new AtomicInteger(0); private final ExpiredOperationReaper expirationReaper; - public DelayedOperationManager( - String managerName, int serverId, int purgeInterval, MetricGroup metricGroup) { + public DelayedOperationManager(String managerName, int serverId, int purgeInterval) { this.managerName = managerName; this.timeoutTimer = new DefaultTimer(managerName); this.serverId = serverId; @@ -78,8 +75,6 @@ public DelayedOperationManager( this.expirationReaper = new ExpiredOperationReaper(); expirationReaper.start(); - - metricGroup.gauge(MetricNames.DELAYED_OPERATIONS_SIZE, this::numDelayed); } /** diff --git a/fluss-server/src/main/java/com/alibaba/fluss/server/replica/delay/DelayedWriteKey.java b/fluss-server/src/main/java/com/alibaba/fluss/server/replica/delay/DelayedTableBucketKey.java similarity index 80% rename from fluss-server/src/main/java/com/alibaba/fluss/server/replica/delay/DelayedWriteKey.java rename to fluss-server/src/main/java/com/alibaba/fluss/server/replica/delay/DelayedTableBucketKey.java index d214e785a9..c0175dc13b 100644 --- a/fluss-server/src/main/java/com/alibaba/fluss/server/replica/delay/DelayedWriteKey.java +++ b/fluss-server/src/main/java/com/alibaba/fluss/server/replica/delay/DelayedTableBucketKey.java @@ -18,11 +18,14 @@ import com.alibaba.fluss.metadata.TableBucket; -/** Delayed write key, used as watch key in delay operation manager. */ -public class DelayedWriteKey implements DelayedOperationKey { +/** + * Delayed table bucket key, used as watch key in delay operation manager, such as {@link + * DelayedWrite} and {@link DelayedFetchLog}. + */ +public class DelayedTableBucketKey implements DelayedOperationKey { private final TableBucket tableBucket; - public DelayedWriteKey(TableBucket tableBucket) { + public DelayedTableBucketKey(TableBucket tableBucket) { this.tableBucket = tableBucket; } @@ -43,7 +46,7 @@ public boolean equals(Object o) { if (o == null || getClass() != o.getClass()) { return false; } - DelayedWriteKey that = (DelayedWriteKey) o; + DelayedTableBucketKey that = (DelayedTableBucketKey) o; return tableBucket.equals(that.tableBucket); } diff --git a/fluss-server/src/main/java/com/alibaba/fluss/server/replica/delay/DelayedWrite.java b/fluss-server/src/main/java/com/alibaba/fluss/server/replica/delay/DelayedWrite.java index eecb7b6b03..58e98b2497 100644 --- a/fluss-server/src/main/java/com/alibaba/fluss/server/replica/delay/DelayedWrite.java +++ b/fluss-server/src/main/java/com/alibaba/fluss/server/replica/delay/DelayedWrite.java @@ -21,6 +21,7 @@ import com.alibaba.fluss.rpc.messages.ProduceLogRequest; import com.alibaba.fluss.rpc.messages.PutKvRequest; import com.alibaba.fluss.rpc.protocol.Errors; +import com.alibaba.fluss.server.metrics.group.TabletServerMetricGroup; import com.alibaba.fluss.server.replica.Replica; import com.alibaba.fluss.server.replica.ReplicaManager; import com.alibaba.fluss.utils.types.Tuple2; @@ -48,16 +49,19 @@ public class DelayedWrite extends DelayedOperati private final DelayedWriteMetadata delayedWriteMetadata; private final ReplicaManager replicaManager; private final Consumer> callback; + private final TabletServerMetricGroup serverMetricGroup; public DelayedWrite( long delayMs, DelayedWriteMetadata delayedWriteMetadata, ReplicaManager replicaManager, - Consumer> callback) { + Consumer> callback, + TabletServerMetricGroup serverMetricGroup) { super(delayMs); this.delayedWriteMetadata = delayedWriteMetadata; this.replicaManager = replicaManager; this.callback = callback; + this.serverMetricGroup = serverMetricGroup; // first update the acks pending variable according to the error code. updateStatus(); @@ -146,6 +150,7 @@ public boolean tryComplete() { @Override public void onExpiration() { + serverMetricGroup.delayedWriteExpireCount().inc(); delayedWriteMetadata .getBucketStatusMap() .forEach( diff --git a/fluss-server/src/main/java/com/alibaba/fluss/server/replica/fetcher/RemoteLeaderEndpoint.java b/fluss-server/src/main/java/com/alibaba/fluss/server/replica/fetcher/RemoteLeaderEndpoint.java index f4f1145e7d..e2eda132e3 100644 --- a/fluss-server/src/main/java/com/alibaba/fluss/server/replica/fetcher/RemoteLeaderEndpoint.java +++ b/fluss-server/src/main/java/com/alibaba/fluss/server/replica/fetcher/RemoteLeaderEndpoint.java @@ -47,6 +47,9 @@ final class RemoteLeaderEndpoint implements LeaderEndpoint { /** The max fetch size for a bucket in bytes. */ private final int maxFetchSizeForBucket; + private final int minFetchBytes; + private final int maxFetchWaitMs; + RemoteLeaderEndpoint( Configuration conf, int followerServerId, @@ -54,9 +57,12 @@ final class RemoteLeaderEndpoint implements LeaderEndpoint { TabletServerGateway tabletServerGateway) { this.followerServerId = followerServerId; this.remoteNode = remoteNode; - this.maxFetchSize = (int) conf.get(ConfigOptions.LOG_FETCH_MAX_BYTES).getBytes(); + this.maxFetchSize = (int) conf.get(ConfigOptions.LOG_REPLICA_FETCH_MAX_BYTES).getBytes(); this.maxFetchSizeForBucket = - (int) conf.get(ConfigOptions.LOG_FETCH_MAX_BYTES_FOR_BUCKET).getBytes(); + (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(); this.tabletServerGateway = tabletServerGateway; } @@ -87,7 +93,12 @@ public CompletableFuture> fetchLog( public Optional buildFetchLogRequest( Map replicas) { return buildFetchLogRequest( - replicas, followerServerId, maxFetchSize, maxFetchSizeForBucket); + replicas, + followerServerId, + maxFetchSize, + maxFetchSizeForBucket, + minFetchBytes, + maxFetchWaitMs); } @Override @@ -99,11 +110,15 @@ static Optional buildFetchLogRequest( Map replicas, int followerServerId, int maxFetchSize, - int maxFetchSizeForBucket) { + int maxFetchSizeForBucket, + int minFetchBytes, + int maxFetchWaitMs) { FetchLogRequest fetchRequest = new FetchLogRequest() .setFollowerServerId(followerServerId) - .setMaxBytes(maxFetchSize); + .setMaxBytes(maxFetchSize) + .setMinBytes(minFetchBytes) + .setMaxWaitMs(maxFetchWaitMs); Map> fetchLogReqForBuckets = new HashMap<>(); int readyForFetchCount = 0; for (Map.Entry entry : replicas.entrySet()) { diff --git a/fluss-server/src/main/java/com/alibaba/fluss/server/replica/fetcher/ReplicaFetcherThread.java b/fluss-server/src/main/java/com/alibaba/fluss/server/replica/fetcher/ReplicaFetcherThread.java index 3212cf1712..d0411f9475 100644 --- a/fluss-server/src/main/java/com/alibaba/fluss/server/replica/fetcher/ReplicaFetcherThread.java +++ b/fluss-server/src/main/java/com/alibaba/fluss/server/replica/fetcher/ReplicaFetcherThread.java @@ -112,6 +112,8 @@ public int getBucketCount() { @Override public void doWork() { maybeFetch(); + // TODO, if we support fetch from follower, we need to complete delayed fetch log operation + // here. } private void maybeFetch() { diff --git a/fluss-server/src/main/java/com/alibaba/fluss/server/tablet/TabletService.java b/fluss-server/src/main/java/com/alibaba/fluss/server/tablet/TabletService.java index cee7658909..ce517e8da1 100644 --- a/fluss-server/src/main/java/com/alibaba/fluss/server/tablet/TabletService.java +++ b/fluss-server/src/main/java/com/alibaba/fluss/server/tablet/TabletService.java @@ -63,6 +63,8 @@ import java.util.Set; import java.util.concurrent.CompletableFuture; +import static com.alibaba.fluss.server.log.FetchParams.DEFAULT_MAX_WAIT_MS_WHEN_MIN_BYTES_ENABLE; +import static com.alibaba.fluss.server.utils.RpcMessageUtils.getFetchLogData; import static com.alibaba.fluss.server.utils.RpcMessageUtils.makeLookupResponse; import static com.alibaba.fluss.server.utils.RpcMessageUtils.makePrefixLookupResponse; import static com.alibaba.fluss.server.utils.RpcMessageUtils.toLookupData; @@ -112,9 +114,20 @@ public CompletableFuture produceLog(ProduceLogRequest reques @Override public CompletableFuture fetchLog(FetchLogRequest request) { CompletableFuture response = new CompletableFuture<>(); - Map fetchLogData = RpcMessageUtils.getFetchLogData(request); - FetchParams fetchParams = - new FetchParams(request.getFollowerServerId(), request.getMaxBytes()); + Map fetchLogData = getFetchLogData(request); + FetchParams fetchParams; + if (request.hasMinBytes()) { + fetchParams = + new FetchParams( + request.getFollowerServerId(), + request.getMaxBytes(), + request.getMinBytes(), + request.hasMaxWaitMs() + ? request.getMaxWaitMs() + : DEFAULT_MAX_WAIT_MS_WHEN_MIN_BYTES_ENABLE); + } else { + fetchParams = new FetchParams(request.getFollowerServerId(), request.getMaxBytes()); + } replicaManager.fetchLogRecords( fetchParams, fetchLogData, diff --git a/fluss-server/src/test/java/com/alibaba/fluss/server/replica/AdjustIsrTest.java b/fluss-server/src/test/java/com/alibaba/fluss/server/replica/AdjustIsrTest.java index 46d333d965..eb2aa7029b 100644 --- a/fluss-server/src/test/java/com/alibaba/fluss/server/replica/AdjustIsrTest.java +++ b/fluss-server/src/test/java/com/alibaba/fluss/server/replica/AdjustIsrTest.java @@ -68,7 +68,8 @@ void testExpandIsr() throws Exception { // mock follower 2 to fetch data from leader. fetch offset is 10 (which indicate the // follower catch up the leader, it will be added into isr list). replicaManager.fetchLogRecords( - new FetchParams(2, (int) conf.get(ConfigOptions.LOG_FETCH_MAX_BYTES).getBytes()), + new FetchParams( + 2, (int) conf.get(ConfigOptions.LOG_REPLICA_FETCH_MAX_BYTES).getBytes()), Collections.singletonMap( tb, new FetchData(tb.getTableId(), 10L, Integer.MAX_VALUE)), result -> {}); @@ -82,7 +83,8 @@ tb, new FetchData(tb.getTableId(), 10L, Integer.MAX_VALUE)), // mock follower 3 to fetch data from leader. fetch offset is 10 (which indicate the // follower catch up the leader, it will be added into isr list). replicaManager.fetchLogRecords( - new FetchParams(3, (int) conf.get(ConfigOptions.LOG_FETCH_MAX_BYTES).getBytes()), + new FetchParams( + 3, (int) conf.get(ConfigOptions.LOG_REPLICA_FETCH_MAX_BYTES).getBytes()), Collections.singletonMap( tb, new FetchData(tb.getTableId(), 10L, Integer.MAX_VALUE)), result -> {}); diff --git a/fluss-server/src/test/java/com/alibaba/fluss/server/replica/ReplicaManagerTest.java b/fluss-server/src/test/java/com/alibaba/fluss/server/replica/ReplicaManagerTest.java index e438ee132a..32990148da 100644 --- a/fluss-server/src/test/java/com/alibaba/fluss/server/replica/ReplicaManagerTest.java +++ b/fluss-server/src/test/java/com/alibaba/fluss/server/replica/ReplicaManagerTest.java @@ -161,6 +161,20 @@ void testFetchLog() throws Exception { TableBucket tb = new TableBucket(DATA1_TABLE_ID, 1); makeLogTableAsLeader(tb.getBucket()); + // test fetch empty buckets without log segment. + CompletableFuture> emptyFuture = + new CompletableFuture<>(); + replicaManager.fetchLogRecords( + buildFetchParams(-1), + Collections.singletonMap(tb, new FetchData(tb.getTableId(), 0L, 1024 * 1024)), + emptyFuture::complete); + Map result = emptyFuture.get(); + assertThat(result.size()).isEqualTo(1); + FetchLogResultForBucket resultForBucket = result.get(tb); + assertThat(resultForBucket.getTableBucket()).isEqualTo(tb); + assertThat(resultForBucket.getHighWatermark()).isEqualTo(0L); + assertThat(resultForBucket.records().sizeInBytes()).isEqualTo(0); + // produce one batch to this bucket. CompletableFuture> future = new CompletableFuture<>(); replicaManager.appendRecordsToLog( @@ -177,9 +191,9 @@ void testFetchLog() throws Exception { buildFetchParams(-1), Collections.singletonMap(tb, new FetchData(tb.getTableId(), 0L, 1024 * 1024)), future1::complete); - Map result = future1.get(); + result = future1.get(); assertThat(result.size()).isEqualTo(1); - FetchLogResultForBucket resultForBucket = result.get(tb); + resultForBucket = result.get(tb); assertThat(resultForBucket.getTableBucket()).isEqualTo(tb); assertThat(resultForBucket.getHighWatermark()).isEqualTo(10L); LogRecords records = resultForBucket.records(); diff --git a/fluss-server/src/test/java/com/alibaba/fluss/server/replica/ReplicaTest.java b/fluss-server/src/test/java/com/alibaba/fluss/server/replica/ReplicaTest.java index d9cf963c4a..1d3e0e4724 100644 --- a/fluss-server/src/test/java/com/alibaba/fluss/server/replica/ReplicaTest.java +++ b/fluss-server/src/test/java/com/alibaba/fluss/server/replica/ReplicaTest.java @@ -113,7 +113,11 @@ void testAppendRecordsToLeader() throws Exception { assertThat(appendInfo.shallowCount()).isEqualTo(1); FetchParams fetchParams = - new FetchParams(-1, (int) conf.get(ConfigOptions.LOG_FETCH_MAX_BYTES).getBytes()); + new FetchParams( + -1, + (int) + conf.get(ConfigOptions.CLIENT_SCANNER_LOG_FETCH_MAX_BYTES) + .getBytes()); fetchParams.setCurrentFetch(DATA1_TABLE_ID, 0, Integer.MAX_VALUE, DATA1_ROW_TYPE, null); LogReadInfo logReadInfo = logReplica.fetchRecords(fetchParams); assertLogRecordsEquals(DATA1_ROW_TYPE, logReadInfo.getFetchedData().getRecords(), DATA1); diff --git a/fluss-server/src/test/java/com/alibaba/fluss/server/replica/ReplicaTestBase.java b/fluss-server/src/test/java/com/alibaba/fluss/server/replica/ReplicaTestBase.java index 2db85ff7c5..286e6cb15f 100644 --- a/fluss-server/src/test/java/com/alibaba/fluss/server/replica/ReplicaTestBase.java +++ b/fluss-server/src/test/java/com/alibaba/fluss/server/replica/ReplicaTestBase.java @@ -427,6 +427,7 @@ private Replica makeReplica( conf.getString(ConfigOptions.DATA_DIR), HIGH_WATERMARK_CHECKPOINT_FILE_NAME))), replicaManager.getDelayedWriteManager(), + replicaManager.getDelayedFetchLogManager(), replicaManager.getAdjustIsrManager(), snapshotContext, serverMetadataCache, diff --git a/fluss-server/src/test/java/com/alibaba/fluss/server/replica/delay/DelayedFetchLogTest.java b/fluss-server/src/test/java/com/alibaba/fluss/server/replica/delay/DelayedFetchLogTest.java new file mode 100644 index 0000000000..782cd59215 --- /dev/null +++ b/fluss-server/src/test/java/com/alibaba/fluss/server/replica/delay/DelayedFetchLogTest.java @@ -0,0 +1,162 @@ +/* + * Copyright (c) 2024 Alibaba Group Holding Ltd. + * + * Licensed 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 com.alibaba.fluss.server.replica.delay; + +import com.alibaba.fluss.metadata.TableBucket; +import com.alibaba.fluss.record.MemoryLogRecords; +import com.alibaba.fluss.rpc.entity.FetchLogResultForBucket; +import com.alibaba.fluss.rpc.entity.ProduceLogResultForBucket; +import com.alibaba.fluss.server.entity.FetchData; +import com.alibaba.fluss.server.log.FetchParams; +import com.alibaba.fluss.server.log.LogOffsetMetadata; +import com.alibaba.fluss.server.metrics.group.TestingMetricGroups; +import com.alibaba.fluss.server.replica.ReplicaTestBase; +import com.alibaba.fluss.server.replica.delay.DelayedFetchLog.FetchBucketStatus; + +import org.junit.jupiter.api.Test; + +import java.time.Duration; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.function.Consumer; + +import static com.alibaba.fluss.record.TestData.DATA1; +import static com.alibaba.fluss.record.TestData.DATA1_ROW_TYPE; +import static com.alibaba.fluss.record.TestData.DATA1_TABLE_ID; +import static com.alibaba.fluss.testutils.DataTestUtils.assertLogRecordsEquals; +import static com.alibaba.fluss.testutils.DataTestUtils.genMemoryLogRecordsByObject; +import static com.alibaba.fluss.testutils.common.CommonTestUtils.retry; +import static org.assertj.core.api.Assertions.assertThat; + +/** Test for {@link DelayedFetchLog}. */ +public class DelayedFetchLogTest extends ReplicaTestBase { + + @Test + void testCompleteDelayedFetchLog() throws Exception { + TableBucket tb = new TableBucket(DATA1_TABLE_ID, 1); + makeLogTableAsLeader(tb.getBucket()); + + FetchLogResultForBucket preFetchResultForBucket = + new FetchLogResultForBucket(tb, MemoryLogRecords.EMPTY, 0L); + CompletableFuture> delayedResponse = + new CompletableFuture<>(); + DelayedFetchLog delayedFetchLog = + createDelayedFetchLogRequest( + tb, + 100, + Duration.ofMinutes(3).toMillis(), // max wait ms large enough. + new FetchBucketStatus( + new FetchData(150001L, 0L, Integer.MAX_VALUE), + new LogOffsetMetadata(0L, 0L, 0), + preFetchResultForBucket), + delayedResponse::complete); + + DelayedOperationManager delayedFetchLogManager = + replicaManager.getDelayedFetchLogManager(); + DelayedTableBucketKey delayedTableBucketKey = new DelayedTableBucketKey(tb); + boolean completed = + delayedFetchLogManager.tryCompleteElseWatch( + delayedFetchLog, Collections.singletonList(delayedTableBucketKey)); + assertThat(completed).isFalse(); + assertThat(delayedFetchLogManager.numDelayed()).isEqualTo(1); + assertThat(delayedFetchLogManager.watched()).isEqualTo(1); + + int numComplete = delayedFetchLogManager.checkAndComplete(delayedTableBucketKey); + assertThat(numComplete).isEqualTo(0); + assertThat(delayedFetchLogManager.numDelayed()).isEqualTo(1); + assertThat(delayedFetchLogManager.watched()).isEqualTo(1); + + // write data. + assertThat(delayedResponse.isDone()).isFalse(); + CompletableFuture> future = new CompletableFuture<>(); + replicaManager.appendRecordsToLog( + 20000, + -1, + Collections.singletonMap(tb, genMemoryLogRecordsByObject(DATA1)), + future::complete); + assertThat(future.get()).containsOnly(new ProduceLogResultForBucket(tb, 0, 10L)); + + assertThat(delayedFetchLogManager.numDelayed()).isEqualTo(0); + assertThat(delayedFetchLogManager.watched()).isEqualTo(0); + + Map result = delayedResponse.get(); + assertThat(result.containsKey(tb)).isTrue(); + FetchLogResultForBucket resultForBucket = result.get(tb); + assertThat(resultForBucket.getHighWatermark()).isEqualTo(10L); + assertLogRecordsEquals(DATA1_ROW_TYPE, resultForBucket.records(), DATA1); + } + + @Test + void testDelayFetchLogTimeout() { + TableBucket tb = new TableBucket(DATA1_TABLE_ID, 1); + makeLogTableAsLeader(tb.getBucket()); + + FetchLogResultForBucket preFetchResultForBucket = + new FetchLogResultForBucket(tb, MemoryLogRecords.EMPTY, 0L); + CompletableFuture> delayedResponse = + new CompletableFuture<>(); + DelayedFetchLog delayedFetchLog = + createDelayedFetchLogRequest( + tb, + 100, + 1000, // wait time is small enough. + new FetchBucketStatus( + new FetchData(150001L, 0L, Integer.MAX_VALUE), + new LogOffsetMetadata(0L, 0L, 0), + preFetchResultForBucket), + delayedResponse::complete); + + DelayedOperationManager delayedFetchLogManager = + replicaManager.getDelayedFetchLogManager(); + DelayedTableBucketKey delayedTableBucketKey = new DelayedTableBucketKey(tb); + boolean completed = + delayedFetchLogManager.tryCompleteElseWatch( + delayedFetchLog, Collections.singletonList(delayedTableBucketKey)); + assertThat(completed).isFalse(); + retry( + Duration.ofMinutes(1), + () -> { + delayedFetchLogManager.checkAndComplete(delayedTableBucketKey); + assertThat(delayedFetchLogManager.numDelayed()).isEqualTo(0); + assertThat(delayedFetchLogManager.watched()).isEqualTo(0); + + assertThat(delayedResponse.isDone()).isTrue(); + Map result = delayedResponse.get(); + assertThat(result.containsKey(tb)).isTrue(); + FetchLogResultForBucket resultForBucket = result.get(tb); + assertThat(resultForBucket.getHighWatermark()).isEqualTo(0L); + assertThat(resultForBucket.recordsOrEmpty()).isEqualTo(MemoryLogRecords.EMPTY); + }); + } + + private DelayedFetchLog createDelayedFetchLogRequest( + TableBucket tb, + int minFetchSize, + long maxWaitMs, + FetchBucketStatus prevFetchBucketStatus, + Consumer> responseCallback) { + FetchParams fetchParams = new FetchParams(-1, Integer.MAX_VALUE, minFetchSize, maxWaitMs); + return new DelayedFetchLog( + fetchParams, + replicaManager, + Collections.singletonMap(tb, prevFetchBucketStatus), + responseCallback, + TestingMetricGroups.TABLET_SERVER_METRICS); + } +} diff --git a/fluss-server/src/test/java/com/alibaba/fluss/server/replica/delay/DelayedOperationTest.java b/fluss-server/src/test/java/com/alibaba/fluss/server/replica/delay/DelayedOperationTest.java index 15d66f1587..6ef14d5710 100644 --- a/fluss-server/src/test/java/com/alibaba/fluss/server/replica/delay/DelayedOperationTest.java +++ b/fluss-server/src/test/java/com/alibaba/fluss/server/replica/delay/DelayedOperationTest.java @@ -17,7 +17,6 @@ package com.alibaba.fluss.server.replica.delay; import com.alibaba.fluss.exception.FlussRuntimeException; -import com.alibaba.fluss.metrics.util.NOPMetricsGroup; import org.assertj.core.api.Assertions; import org.junit.jupiter.api.AfterEach; @@ -46,8 +45,7 @@ public class DelayedOperationTest { @BeforeEach void setup() { - delayedOperationManager = - new DelayedOperationManager<>("test", 0, 1000, NOPMetricsGroup.newInstance()); + delayedOperationManager = new DelayedOperationManager<>("test", 0, 1000); } @AfterEach diff --git a/fluss-server/src/test/java/com/alibaba/fluss/server/replica/delay/DelayedWriteTest.java b/fluss-server/src/test/java/com/alibaba/fluss/server/replica/delay/DelayedWriteTest.java index b7aab4b34d..c949a9a272 100644 --- a/fluss-server/src/test/java/com/alibaba/fluss/server/replica/delay/DelayedWriteTest.java +++ b/fluss-server/src/test/java/com/alibaba/fluss/server/replica/delay/DelayedWriteTest.java @@ -20,6 +20,7 @@ import com.alibaba.fluss.rpc.entity.ProduceLogResultForBucket; import com.alibaba.fluss.rpc.protocol.Errors; import com.alibaba.fluss.server.log.LogAppendInfo; +import com.alibaba.fluss.server.metrics.group.TestingMetricGroups; import com.alibaba.fluss.server.replica.Replica; import com.alibaba.fluss.server.replica.ReplicaTestBase; import com.alibaba.fluss.server.replica.delay.DelayedWrite.DelayedBucketStatus; @@ -62,22 +63,22 @@ void testCompleteDelayWrite() throws Exception { DelayedOperationManager> delayedWriteManager = replicaManager.getDelayedWriteManager(); - DelayedWriteKey delayedWriteKey = new DelayedWriteKey(tb); + DelayedTableBucketKey delayedTableBucketKey = new DelayedTableBucketKey(tb); boolean completed = delayedWriteManager.tryCompleteElseWatch( - delayedWrite, Collections.singletonList(delayedWriteKey)); + delayedWrite, Collections.singletonList(delayedTableBucketKey)); assertThat(completed).isFalse(); assertThat(delayedWriteManager.numDelayed()).isEqualTo(1); assertThat(delayedWriteManager.watched()).isEqualTo(1); - int numComplete = delayedWriteManager.checkAndComplete(delayedWriteKey); + int numComplete = delayedWriteManager.checkAndComplete(delayedTableBucketKey); assertThat(numComplete).isEqualTo(0); assertThat(delayedWriteManager.numDelayed()).isEqualTo(1); assertThat(delayedWriteManager.watched()).isEqualTo(1); // update highWatermark to log end offset. replica.getLogTablet().updateHighWatermark(10L); - numComplete = delayedWriteManager.checkAndComplete(delayedWriteKey); + numComplete = delayedWriteManager.checkAndComplete(delayedTableBucketKey); assertThat(numComplete).isEqualTo(1); assertThat(delayedWriteManager.numDelayed()).isEqualTo(0); assertThat(delayedWriteManager.watched()).isEqualTo(0); @@ -89,8 +90,8 @@ void testDelayedWriteKey() throws Exception { makeLogTableAsLeader(tb.getBucket()); Replica replica = replicaManager.getReplicaOrException(tb); - DelayedWriteKey delayedWriteKey = new DelayedWriteKey(tb); - assertThat(delayedWriteKey.getTableBucket()).isEqualTo(tb); + DelayedTableBucketKey delayedTableBucketKey = new DelayedTableBucketKey(tb); + assertThat(delayedTableBucketKey.getTableBucket()).isEqualTo(tb); DelayedOperationManager> delayedWriteManager = replicaManager.getDelayedWriteManager(); @@ -104,20 +105,20 @@ void testDelayedWriteKey() throws Exception { assertThat(resultList.get(0).getWriteLogEndOffset()).isEqualTo(10); }); // Directly watch for the delayedWrite. - delayedWriteManager.watchForOperation(delayedWriteKey, delayedWrite); + delayedWriteManager.watchForOperation(delayedTableBucketKey, delayedWrite); assertThat(delayedWriteManager.watched()).isEqualTo(1); // new a delayedWriteKey point to the same tableBucket. It's import to new a delayedWriteKey // because when we want to check and complete the operation in ReplicaManager/Replica, // we will always make a new delayedWriteKey. - DelayedWriteKey newDelayedWriteKey = new DelayedWriteKey(tb); - int numComplete = delayedWriteManager.checkAndComplete(newDelayedWriteKey); + DelayedTableBucketKey newDelayedTableBucketKey = new DelayedTableBucketKey(tb); + int numComplete = delayedWriteManager.checkAndComplete(newDelayedTableBucketKey); assertThat(numComplete).isEqualTo(0); assertThat(delayedWriteManager.watched()).isEqualTo(1); // update highWatermark to log end offset. replica.getLogTablet().updateHighWatermark(10L); - numComplete = delayedWriteManager.checkAndComplete(newDelayedWriteKey); + numComplete = delayedWriteManager.checkAndComplete(newDelayedTableBucketKey); assertThat(numComplete).isEqualTo(1); assertThat(delayedWriteManager.watched()).isEqualTo(0); } @@ -142,15 +143,15 @@ void testDelayWriteTimeOut() throws Exception { DelayedOperationManager> delayedWriteManager = replicaManager.getDelayedWriteManager(); - DelayedWriteKey delayedWriteKey = new DelayedWriteKey(tb); + DelayedTableBucketKey delayedTableBucketKey = new DelayedTableBucketKey(tb); boolean completed = delayedWriteManager.tryCompleteElseWatch( - delayedWrite, Collections.singletonList(delayedWriteKey)); + delayedWrite, Collections.singletonList(delayedTableBucketKey)); assertThat(completed).isFalse(); assertThat(delayedWriteManager.numDelayed()).isEqualTo(1); assertThat(delayedWriteManager.watched()).isEqualTo(1); - int numComplete = delayedWriteManager.checkAndComplete(delayedWriteKey); + int numComplete = delayedWriteManager.checkAndComplete(delayedTableBucketKey); assertThat(numComplete).isEqualTo(0); assertThat(delayedWriteManager.numDelayed()).isEqualTo(1); assertThat(delayedWriteManager.watched()).isEqualTo(1); @@ -178,6 +179,10 @@ private DelayedWrite createDelayedWrite( Collections.singletonMap(tb, new DelayedBucketStatus<>(10, appendResult)); return new DelayedWrite<>( - delayMs, new DelayedWriteMetadata<>(-1, bucketStatusMap), replicaManager, callback); + delayMs, + new DelayedWriteMetadata<>(-1, bucketStatusMap), + replicaManager, + callback, + TestingMetricGroups.TABLET_SERVER_METRICS); } } diff --git a/fluss-server/src/test/java/com/alibaba/fluss/server/replica/fetcher/ReplicaFetcherITCase.java b/fluss-server/src/test/java/com/alibaba/fluss/server/replica/fetcher/ReplicaFetcherITCase.java index 7d2bb114b7..395dce4ed3 100644 --- a/fluss-server/src/test/java/com/alibaba/fluss/server/replica/fetcher/ReplicaFetcherITCase.java +++ b/fluss-server/src/test/java/com/alibaba/fluss/server/replica/fetcher/ReplicaFetcherITCase.java @@ -59,7 +59,9 @@ import static com.alibaba.fluss.record.TestData.DATA_1_WITH_KEY_AND_VALUE; import static com.alibaba.fluss.record.TestData.EXPECTED_LOG_RESULTS_FOR_DATA_1_WITH_PK; import static com.alibaba.fluss.server.testutils.KvTestUtils.assertLookupResponse; +import static com.alibaba.fluss.server.testutils.RpcMessageTestUtils.assertFetchLogResponse; import static com.alibaba.fluss.server.testutils.RpcMessageTestUtils.assertFetchLogResponseWithRowKind; +import static com.alibaba.fluss.server.testutils.RpcMessageTestUtils.assertProduceLogResponse; import static com.alibaba.fluss.server.testutils.RpcMessageTestUtils.createTable; import static com.alibaba.fluss.server.testutils.RpcMessageTestUtils.newFetchLogRequest; import static com.alibaba.fluss.server.testutils.RpcMessageTestUtils.newLookupRequest; @@ -121,7 +123,7 @@ void testProduceLogNeedAck() throws Exception { FLUSS_CLUSTER_EXTENSION.newTabletServerClientForNode(leader); // send one batch, which need ack. - RpcMessageTestUtils.assertProduceLogResponse( + assertProduceLogResponse( leaderGateWay .produceLog( RpcMessageTestUtils.newProduceLogRequest( @@ -134,7 +136,7 @@ void testProduceLogNeedAck() throws Exception { 0L); // check leader log data. - RpcMessageTestUtils.assertFetchLogResponse( + assertFetchLogResponse( leaderGateWay.fetchLog(newFetchLogRequest(-1, tableId, bucketId, 0L)).get(), tableId, bucketId, @@ -147,13 +149,25 @@ void testProduceLogNeedAck() throws Exception { leaderAndIsr.isr().stream() .filter(id -> id != leader) .collect(Collectors.toList())) { + ReplicaManager replicaManager = FLUSS_CLUSTER_EXTENSION.getTabletServerById(followId).getReplicaManager(); + // wait util follower highWaterMark equals leader. + retry( + Duration.ofMinutes(1), + () -> + assertThat( + replicaManager + .getReplicaOrException(tb) + .getLogTablet() + .getHighWatermark()) + .isEqualTo(10L)); + CompletableFuture> future = new CompletableFuture<>(); // mock client fetch from follower. replicaManager.fetchLogRecords( - new FetchParams(-1, false, Integer.MAX_VALUE), + new FetchParams(-1, false, Integer.MAX_VALUE, -1, -1), Collections.singletonMap(tb, new FetchData(tableId, 0L, 1024 * 1024)), future::complete); Map result = future.get(); @@ -163,17 +177,6 @@ void testProduceLogNeedAck() throws Exception { LogRecords records = resultForBucket.records(); assertThat(records).isNotNull(); assertLogRecordsEquals(DATA1_ROW_TYPE, records, DATA1); - - // wait util follower highWaterMark equals leader. - retry( - Duration.ofMinutes(1), - () -> - assertThat( - replicaManager - .getReplicaOrException(tb) - .getLogTablet() - .getHighWatermark()) - .isEqualTo(10L)); } } @@ -245,7 +248,7 @@ void testPutKvNeedAck() throws Exception { new CompletableFuture<>(); // mock client fetch from follower. replicaManager.fetchLogRecords( - new FetchParams(-1, false, Integer.MAX_VALUE), + new FetchParams(-1, false, Integer.MAX_VALUE, -1, -1), Collections.singletonMap(tb, new FetchData(tableId, 0L, 1024 * 1024)), future::complete); Map result = future.get(); diff --git a/fluss-server/src/test/java/com/alibaba/fluss/server/replica/fetcher/TestingLeaderEndpoint.java b/fluss-server/src/test/java/com/alibaba/fluss/server/replica/fetcher/TestingLeaderEndpoint.java index 3318cd5e93..293fdbfdb2 100644 --- a/fluss-server/src/test/java/com/alibaba/fluss/server/replica/fetcher/TestingLeaderEndpoint.java +++ b/fluss-server/src/test/java/com/alibaba/fluss/server/replica/fetcher/TestingLeaderEndpoint.java @@ -53,9 +53,9 @@ public TestingLeaderEndpoint( Configuration conf, ReplicaManager replicaManager, ServerNode localNode) { this.replicaManager = replicaManager; this.localNode = localNode; - this.maxFetchSize = (int) conf.get(ConfigOptions.LOG_FETCH_MAX_BYTES).getBytes(); + this.maxFetchSize = (int) conf.get(ConfigOptions.LOG_REPLICA_FETCH_MAX_BYTES).getBytes(); this.maxFetchSizeForBucket = - (int) conf.get(ConfigOptions.LOG_FETCH_MAX_BYTES_FOR_BUCKET).getBytes(); + (int) conf.get(ConfigOptions.LOG_REPLICA_FETCH_MAX_BYTES_FOR_BUCKET).getBytes(); } @Override @@ -93,7 +93,7 @@ public CompletableFuture> fetchLog( public Optional buildFetchLogRequest( Map replicas) { return RemoteLeaderEndpoint.buildFetchLogRequest( - replicas, localNode.id(), maxFetchSize, maxFetchSizeForBucket); + replicas, localNode.id(), maxFetchSize, maxFetchSizeForBucket, -1, -1); } @Override diff --git a/fluss-server/src/test/java/com/alibaba/fluss/server/tablet/TabletServiceITCase.java b/fluss-server/src/test/java/com/alibaba/fluss/server/tablet/TabletServiceITCase.java index 24c6e97f63..b3d5afdcd3 100644 --- a/fluss-server/src/test/java/com/alibaba/fluss/server/tablet/TabletServiceITCase.java +++ b/fluss-server/src/test/java/com/alibaba/fluss/server/tablet/TabletServiceITCase.java @@ -27,9 +27,12 @@ import com.alibaba.fluss.row.encode.KeyEncoder; import com.alibaba.fluss.row.encode.ValueEncoder; import com.alibaba.fluss.rpc.gateway.TabletServerGateway; +import com.alibaba.fluss.rpc.messages.FetchLogResponse; import com.alibaba.fluss.rpc.messages.InitWriterRequest; import com.alibaba.fluss.rpc.messages.InitWriterResponse; import com.alibaba.fluss.rpc.messages.ListOffsetsResponse; +import com.alibaba.fluss.rpc.messages.PbFetchLogRespForBucket; +import com.alibaba.fluss.rpc.messages.PbFetchLogRespForTable; import com.alibaba.fluss.rpc.messages.PbListOffsetsRespForBucket; import com.alibaba.fluss.rpc.messages.PbLookupRespForBucket; import com.alibaba.fluss.rpc.messages.PbPrefixLookupRespForBucket; @@ -48,10 +51,12 @@ import javax.annotation.Nullable; +import java.time.Duration; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.List; +import java.util.concurrent.CompletableFuture; import static com.alibaba.fluss.record.TestData.ANOTHER_DATA1; import static com.alibaba.fluss.record.TestData.DATA1; @@ -249,6 +254,96 @@ void testFetchLog() throws Exception { "Projected fields [2, 3] is out of bound for schema with 2 fields."); } + @Test + void testFetchLogWithMinFetchSizeAndTimeout() throws Exception { + long tableId = + createTable( + FLUSS_CLUSTER_EXTENSION, + DATA1_TABLE_PATH, + DATA1_TABLE_INFO.getTableDescriptor()); + TableBucket tb = new TableBucket(tableId, 0); + + FLUSS_CLUSTER_EXTENSION.waitUtilAllReplicaReady(tb); + + int leader = FLUSS_CLUSTER_EXTENSION.waitAndGetLeader(tb); + TabletServerGateway leaderGateWay = + FLUSS_CLUSTER_EXTENSION.newTabletServerClientForNode(leader); + + // first send an empty fetch request, without min fetch size, the request will return + // immediately. + FetchLogResponse fetchLogResponse = + leaderGateWay.fetchLog(newFetchLogRequest(-1, tableId, 0, 0L)).get(); + assertThat(fetchLogResponse.getTablesRespsCount()).isEqualTo(1); + PbFetchLogRespForTable fetchLogRespForTable = fetchLogResponse.getTablesRespsList().get(0); + assertThat(fetchLogRespForTable.getTableId()).isEqualTo(tableId); + assertThat(fetchLogRespForTable.getBucketsRespsCount()).isEqualTo(1); + PbFetchLogRespForBucket protoFetchedBucket = + fetchLogRespForTable.getBucketsRespsList().get(0); + assertThat(protoFetchedBucket.getHighWatermark()).isEqualTo(0L); + assertThat(protoFetchedBucket.getRecordsSize()).isEqualTo(0); + + // second send a fetch request with minFetchSize and small maxFetchWaitMs, the request will + // also return immediately. + fetchLogResponse = + leaderGateWay + .fetchLog( + newFetchLogRequest( + -1, tableId, 0, 0L, null, 1, Integer.MAX_VALUE, 100)) + .get(); + assertThat(fetchLogResponse.getTablesRespsCount()).isEqualTo(1); + fetchLogRespForTable = fetchLogResponse.getTablesRespsList().get(0); + assertThat(fetchLogRespForTable.getTableId()).isEqualTo(tableId); + assertThat(fetchLogRespForTable.getBucketsRespsCount()).isEqualTo(1); + protoFetchedBucket = fetchLogRespForTable.getBucketsRespsList().get(0); + assertThat(protoFetchedBucket.getHighWatermark()).isEqualTo(0L); + assertThat(protoFetchedBucket.getRecordsSize()).isEqualTo(0); + + // third send a fetch request with minFetchSize and much bigger maxFetchWaitMs, the request + // will return after we send a produce log request to this bucket. + CompletableFuture fetchResultFuture = + leaderGateWay.fetchLog( + newFetchLogRequest( + -1, + tableId, + 0, + 0L, + null, + 1, + Integer.MAX_VALUE, + (int) Duration.ofMinutes(5).toMillis())); + + // send a produce log request to trigger delay fetch log finish. + assertProduceLogResponse( + leaderGateWay + .produceLog( + newProduceLogRequest( + tableId, 0, -1, genMemoryLogRecordsByObject(DATA1))) + .get(), + 0, + 0L); + // the delay fetch will be completed. + assertFetchLogResponse(fetchResultFuture.get(), tableId, 0, 10L, DATA1); + + // return immediately. + assertFetchLogResponse( + leaderGateWay + .fetchLog( + newFetchLogRequest( + -1, + tableId, + 0, + 0L, + null, + 1, + Integer.MAX_VALUE, + (int) Duration.ofMinutes(5).toMillis())) + .get(), + tableId, + 0, + 10L, + DATA1); + } + @Test void testInvalidFetchLog() throws Exception { long tableId = diff --git a/fluss-server/src/test/java/com/alibaba/fluss/server/testutils/RpcMessageTestUtils.java b/fluss-server/src/test/java/com/alibaba/fluss/server/testutils/RpcMessageTestUtils.java index e6716a8f6a..296194909f 100644 --- a/fluss-server/src/test/java/com/alibaba/fluss/server/testutils/RpcMessageTestUtils.java +++ b/fluss-server/src/test/java/com/alibaba/fluss/server/testutils/RpcMessageTestUtils.java @@ -187,10 +187,32 @@ public static FetchLogRequest newFetchLogRequest( public static FetchLogRequest newFetchLogRequest( int followerId, long tableId, int bucketId, long fetchOffset, int[] selectedFields) { + return newFetchLogRequest( + followerId, + tableId, + bucketId, + fetchOffset, + selectedFields, + -1, + Integer.MAX_VALUE, + -1); + } + + public static FetchLogRequest newFetchLogRequest( + int followerId, + long tableId, + int bucketId, + long fetchOffset, + int[] selectedFields, + int minFetchBytes, + int maxFetchBytes, + int maxWaitMs) { FetchLogRequest fetchLogRequest = - new FetchLogRequest() - .setFollowerServerId(followerId) - .setMaxBytes(Integer.MAX_VALUE); + new FetchLogRequest().setFollowerServerId(followerId).setMaxBytes(maxFetchBytes); + if (minFetchBytes > 0) { + fetchLogRequest.setMinBytes(minFetchBytes).setMaxWaitMs(maxWaitMs); + } + PbFetchLogReqForTable fetchLogReqForTable = new PbFetchLogReqForTable().setTableId(tableId); if (selectedFields != null) { fetchLogReqForTable diff --git a/website/docs/maintenance/configuration.md b/website/docs/maintenance/configuration.md index e82ba62c59..a5dd979051 100644 --- a/website/docs/maintenance/configuration.md +++ b/website/docs/maintenance/configuration.md @@ -96,10 +96,13 @@ during the Fluss cluster working. | log.replica.high-watermark.checkpoint-interval | Duration | 5s | The frequency with which the high watermark is saved out to disk. The default setting is 5 seconds. | | log.replica.max-lag-time | Duration | 30s | If a follower replica hasn't sent any fetch log requests or hasn't consumed up the leaders log end offset for at least this time, the leader will remove the follower replica form isr | | log.replica.write-operation-purge-number | Integer | 1000 | The purge number (in number of requests) of the write operation manager, the default value is 1000. | +| log.replica.fetch-operation-purge-number | Integer | 1000 | The purge number (in number of requests) of the fetch log operation manager, the default value is 1000. | | log.replica.fetcher-number | Integer | 1 | Number of fetcher threads used to replicate log records from each source tablet server. The total number of fetchers on each tablet server is bound by this parameter multiplied by the number of tablet servers in the cluster. Increasing this value can increase the degree of I/O parallelism in the follower and leader tablet server at the cost of higher CPU and memory utilization. | -| log.replica.fetch-backoff-interval | Duration | 1s | The amount of time to sleep when fetch bucket error occurs. | -| log.fetch.max-bytes | MemorySize | 16mb | The maximum amount of data the server should return for a fetch request. Records are fetched in batches for log scanner or follower, for one request batch, and if the first record batch in the first non-empty bucket of the fetch is larger than this value, the record batch will still be returned to ensure that the fetch can make progress. As such, this is not a absolute maximum. Note that the fetcher performs multiple fetches in parallel. | -| log.fetch.max-bytes-for-bucket | MemorySize | 1mb | The maximum amount of data the server should return for a table bucket in fetch request. Records are fetched in batches for consumer or follower, for one request batch, the max bytes size is config by this option. | +| log.replica.fetch.backoff-interval | Duration | 1s | The amount of time to sleep when fetch bucket error occurs. | +| log.replica.fetch.max-bytes | MemorySize | 16mb | The maximum amount of data the server should return for a fetch request from follower. Records are fetched in batches, and if the first record batch in the first non-empty bucket of the fetch is larger than this value, the record batch will still be returned to ensure that the fetch can make progress. As such, this is not a absolute maximum. Note that the fetcher performs multiple fetches in parallel. | +| log.replica.fetch.max-bytes-for-bucket | MemorySize | 1mb | The maximum amount of data the server should return for a table bucket in fetch request fom follower. Records are fetched in batches, and the max bytes size is config by this option. | +| log.replica.fetch.min-bytes | MemorySize | 1b | The minimum bytes expected for each fetch log request from the follower to response. If not enough bytes, wait up to log.replica.fetch-wait-max-time time to return. | +| log.replica.fetch.wait-max-time | Duration | 500ms | The maximum time to wait for enough bytes to be available for a fetch log request from the follower to response. This value should always be less than the 'log.replica.max-lag-time' at all times to prevent frequent shrinking of ISR for low throughput tables | | log.replica.min-in-sync-replicas-number | Integer | 1 | When a producer set acks to all (-1), this configuration specifies the minimum number of replicas that must acknowledge a write for the write to be considered successful. If this minimum cannot be met, then the producer will raise an exception(NotEnoughReplicas). when used together, this config and 'acks' allow you to enforce greater durability guarantees. A typical scenario would be to create a table with a replication factor of 3. set this conf to 2, and produce with acks = -1. This will ensure that the producer raises an exception if a majority of replicas don't receive a write. | ### Log Tiered Storage @@ -216,6 +219,10 @@ Currently, we don't support alter table configuration by Flink. This will be sup | client.request-timeout | Duration | 30s | The timeout for a request to complete. If user set the write ack to -1, this timeout is the max time that delayed write try to complete. The default setting is 30 seconds. | | client.scanner.log.check-crc | Boolean | true | Automatically check the CRC3 of the read records for LogScanner. This ensures no on-the-wire or on-disk corruption to the messages occurred. This check adds some overhead, so it may be disabled in cases seeking extreme performance. | | client.scanner.log.max-poll-records | Integer | 500 | The maximum number of records returned in a single call to poll() for LogScanner. Note that this config doesn't impact the underlying fetching behavior. The Scanner will cache the records from each fetch request and returns them incrementally from each poll. | +| client.scanner.log.fetch.max-bytes | MemorySize | 16mb | The maximum amount of data the server should return for a fetch request from client. Records are fetched in batches, and if the first record batch in the first non-empty bucket of the fetch is larger than this value, the record batch will still be returned to ensure that the fetch can make progress. As such, this is not a absolute maximum. | +| client.scanner.log.fetch.max-bytes-for-bucket | MemorySize | 1mb | The maximum amount of data the server should return for a table bucket in fetch request fom client. Records are fetched in batches, and the max bytes size is config by this option. | +| client.scanner.log.fetch.min-bytes | MemorySize | 1b | The minimum bytes expected for each fetch log request from client to response. If not enough bytes, wait up to client.scanner.log.fetch-wait-max-time time to return. | +| client.scanner.log.fetch.wait-max-time | Duration | 500ms | The maximum time to wait for enough bytes to be available for a fetch log request from client to response. | | client.lookup.queue-size | Integer | 25600 | The maximum number of pending lookup operations. | | client.lookup.max-batch-size | Integer | 128 | The maximum batch size of merging lookup operations to one lookup request. | | client.lookup.max-inflight-requests | Integer | 128 | The maximum number of unacknowledged lookup requests for lookup operations. | diff --git a/website/docs/maintenance/monitor-metrics.md b/website/docs/maintenance/monitor-metrics.md index 7304a7e2ef..795d02b2d6 100644 --- a/website/docs/maintenance/monitor-metrics.md +++ b/website/docs/maintenance/monitor-metrics.md @@ -299,7 +299,7 @@ Some metrics might not be exposed when using other JVM implementations (e.g. IBM - tabletserver + tabletserver - replicationBytesInPerSecond The bytes of data write into follower replica for data sync. @@ -326,10 +326,30 @@ Some metrics might not be exposed when using other JVM implementations (e.g. IBM Gauge - delayedOperationsSize - The delayed operations size in this TabletServer. + delayedWriteCount + The delayed write count in this TabletServer. Gauge + + delayedFetchCount + The delayed fetch log operation count in this TabletServer. + Gauge + + + delayedWriteExpiresPerSecond + The delayed write operation expire count per second in this TabletServer. + Meter + + + delayedFetchFromFollowerExpiresPerSecond + The delayed fetch log operation from follower expire count per second in this TabletServer. + Meter + + + delayedFetchFromClientExpiresPerSecond + The delayed fetch log operation from client expire count per second in this TabletServer. + Meter +