yujun777 commented on code in PR #69831: URL: https://github.com/apache/doris/pull/69831#discussion_r4421989926
########## fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmStreamKeepAliveDaemon.java: ########## @@ -0,0 +1,100 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.mtmv.ivm; + +import org.apache.doris.catalog.Database; +import org.apache.doris.catalog.Env; +import org.apache.doris.catalog.MTMV; +import org.apache.doris.catalog.Table; +import org.apache.doris.catalog.stream.BaseTableStream; +import org.apache.doris.catalog.stream.OlapTableStream; +import org.apache.doris.catalog.stream.TableStreamManager; +import org.apache.doris.common.Config; +import org.apache.doris.common.util.MasterDaemon; +import org.apache.doris.persist.EditLog.EditLogItem; + +import com.google.common.annotations.VisibleForTesting; + +import java.util.ArrayList; +import java.util.List; + +/** + * Keeps the offsets of the idle partitions of IVM streams from expiring with changelog.ttl_seconds (F-2). + * + * <p>An offset moves only when a refresh consumes the partition, and a refresh consumes only the partitions with + * new data, so the offset of a partition nobody writes stays at its last change. Once the partition has been idle + * for longer than the TTL, the next write would make the refresh after it take the offset for expired and rebuild + * the whole MV, though nothing the refresh needs is gone. Each round records a keep-alive TSO for such partitions + * ({@link TableStreamManager#keepAliveIdlePartitions} stream by stream, and + * {@link TableStreamManager#keepAliveIdleCloudPartitions} for all of them in cloud mode, where the offsets live in + * MetaService), which the expiration check of a read takes in place of an older offset. Only IVM streams are kept + * alive; a user stream is left to its consumer. + */ +public class IvmStreamKeepAliveDaemon extends MasterDaemon { + // The daemon checks every second whether ivm_stream_keepalive_interval_second has passed since its last round, + // so that a change of the interval applies at once: a daemon sleeping its own interval would only see it once + // the old interval is over, and a sleep may not be cut short while an edit log is being written. + private static final long CHECK_INTERVAL_MS = 1000L; + + // only read and written by the daemon thread + private long lastRoundMs = 0; + + public IvmStreamKeepAliveDaemon() { + super("ivm-stream-keepalive", CHECK_INTERVAL_MS); + } + + @Override + protected void runAfterCatalogReady() { + if (!Config.enable_table_stream || !Config.enable_ivm_stream_keepalive) { + return; + } + long now = System.currentTimeMillis(); + if (now - lastRoundMs < Config.ivm_stream_keepalive_interval_second * 1000L) { + return; + } + lastRoundMs = now; Review Comment: Confirmed. The ordering makes this real rather than theoretical: `Env.startMasterOnlyDaemonThreads()` starts this daemon before `tsoService`, and `MasterDaemon.runOneCycle()` only waits for `Env.isReady()`, which does not include TSO calibration. The first cycle after an FE restart or a leadership change can therefore run while the service is still uncalibrated: local `getTSO()` throws ("TSO timestamp is not calibrated"), and in cloud `getStatusSnapshot().isInitialized()` is false. `runAfterCatalogReady()` assigns `lastRoundMs = now` before `keepAliveIdlePartitions()`, so that skipped cycle suppresses the next attempt for the whole `ivm_stream_keepalive_interval_second`, and `Daemon.run()` swallows the throwable and keeps the thread alive, which is what makes it silent. The impact is bounded but not zero: the lost round only matters when an offset is already inside its renewal margin and the remaining TTL is shorter than the interval. With the default 24h TTL that is 12h of margin against 600s, so it is harmless; it bites a table with a short `changelog.ttl_seconds`, where the first refresh after the restart fails as strict INCREMENTAL or falls back to COMPLETE although the keep-alive would have covered it. Suggested fix, so that a round the service could not run counts as "not run", and readiness is checked before any lock is taken: ```java long now = System.currentTimeMillis(); if (now - lastRoundMs < Config.ivm_stream_keepalive_interval_second * 1000L) { return; } if (!Env.getCurrentEnv().getTSOService().getStatusSnapshot().isInitialized()) { return; // do not advance lastRoundMs: retry on the next one-second cycle } lastRoundMs = now; keepAliveIdlePartitions(); ``` Gating up front also avoids two consequences of taking the TSO mid-round: `getTSO()` retries with sleeps (200ms per attempt, or `tso_service_update_interval_ms`) while `keepAliveIdlePartitions` holds the base-table read lock, so a failure there stalls writers on that table for up to `tso_max_get_retry_count` sleeps; and one stream's throw aborts the round for every stream after it, because the loop in `IvmStreamKeepAliveDaemon.keepAliveIdlePartitions()` has no per-stream guard. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
