morningman commented on code in PR #69831: URL: https://github.com/apache/doris/pull/69831#discussion_r4421739708
########## 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. `startMasterOnlyDaemonThreads()` starts this daemon before `tsoService`, both leave the readiness wait together, and the TSO service calibrates itself in its own first cycle: it logs the next allocation window and, in cloud mode, advances MetaService's TSO fence. A first check that came before that set `lastRoundMs` and then lost the round (in shared-nothing mode `getTSO()` threw at the first stream with a partition due for renewal, in cloud mode the uninitialized status skipped it), so the next round came a whole `ivm_stream_keepalive_interval_second` later. With the default TTL of a day that delay is harmless, but with a `changelog.ttl_seconds` close to twice the interval, which the config description allows, an idle offset could expire in between: an AUTO refresh falls back to COMPLETE and a strict INCREMENTAL refresh fails. No row is read wrong. Fixed in cf84aba1110: `runAfterCatalogReady` now returns before it sets `lastRoundMs` while `TSOService.getStatusSnapshot().isInitialized()` is false, so the first one-second check after calibration starts the round. `IvmStreamKeepAliveDaemonTest` covers it: no round while the TSO service is not initialized, a round at the first check after it is, and none again before the interval; it fails without the check. -- 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]
