morningman commented on code in PR #69831: URL: https://github.com/apache/doris/pull/69831#discussion_r4423902487
########## 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: Thanks. That gate is what cf84aba1110 added, checked before the interval so that the first one-second check after calibration starts the round; `IvmStreamKeepAliveDaemonTest` covers it. One small correction on "silent": `Daemon.run()` logs the throwable at ERROR before it carries on, so the lost round did show in fe.log. On the two other points, at the current head: - `getTSO()` does not sleep while the service is uncalibrated: it throws before its retry loop. Once calibrated, the only sleep a round of this Master daemon can meet is a logical counter overflow within one physical tick, and then every commit runs the same loop inside the table write lock, so the writers already wait on their own TSOs; a round takes one TSO per stream. K cannot move out of the base table read lock either: a commit takes its TSO inside the write lock, and that is what makes every commit the partition has not had yet get a larger TSO than K. - With the gate, the only call on the per-stream path that throws is `getTSO()`, and its failures (disabled, uncalibrated, retries exhausted) are failures of the whole TSO service, so every stream after it would fail the same way; a per-stream catch would turn one ERROR into one per stream. Anything else that throws there is a bug we want to see rather than step over. A lost round is left to the next one, and the config description now counts that against the margin (00befb02e5e). -- 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]
