This is an automated email from the ASF dual-hosted git repository. pnowojski pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit f93cac36ff51e342a002126487484cc2d3d8a0c9 Author: David Moravek <d...@apache.org> AuthorDate: Wed Nov 15 14:31:44 2023 +0100 [FLINK-20217][task] Allow certains operators to yield to unaligned checkpoint in case timers are firing. Co-authored-by: Piotr Nowojski <piotr.nowoj...@gmail.com> --- .../flink/configuration/CheckpointingOptions.java | 8 + .../api/environment/CheckpointConfig.java | 13 ++ .../flink/streaming/api/graph/StreamConfig.java | 8 + .../api/graph/StreamingJobGraphGenerator.java | 2 + .../api/operators/AbstractStreamOperator.java | 57 ++++- .../api/operators/InternalTimeServiceManager.java | 19 ++ .../operators/InternalTimeServiceManagerImpl.java | 11 + .../api/operators/InternalTimerServiceImpl.java | 26 ++- .../api/operators/MailboxWatermarkProcessor.java | 92 +++++++++ .../api/operators/StreamOperatorFactoryUtil.java | 2 +- .../BatchExecutionInternalTimeServiceManager.java | 7 + .../flink/streaming/runtime/tasks/StreamTask.java | 9 + .../runtime/tasks/mailbox/MailboxProcessor.java | 16 ++ .../CheckpointConfigFromConfigurationTest.java | 8 + .../operators/MailboxWatermarkProcessorTest.java | 120 +++++++++++ ...nalignedCheckpointsInterruptibleTimersTest.java | 230 +++++++++++++++++++++ .../io/checkpointing/UnalignedCheckpointsTest.java | 4 +- .../runtime/operators/TableStreamOperator.java | 11 +- .../streaming/util/TestStreamEnvironment.java | 2 + 19 files changed, 632 insertions(+), 13 deletions(-) diff --git a/flink-core/src/main/java/org/apache/flink/configuration/CheckpointingOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/CheckpointingOptions.java index 10344a3d30c..ec7b82ecfd8 100644 --- a/flink-core/src/main/java/org/apache/flink/configuration/CheckpointingOptions.java +++ b/flink-core/src/main/java/org/apache/flink/configuration/CheckpointingOptions.java @@ -632,6 +632,14 @@ public class CheckpointingOptions { "Forces unaligned checkpoints, particularly allowing them for iterative jobs.") .build()); + @Experimental + public static final ConfigOption<Boolean> ENABLE_UNALIGNED_INTERRUPTIBLE_TIMERS = + ConfigOptions.key("execution.checkpointing.unaligned.interruptible-timers.enabled") + .booleanType() + .defaultValue(false) + .withDescription( + "Allows unaligned checkpoints to skip timers that are currently being fired."); + public static final ConfigOption<Boolean> ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH = ConfigOptions.key("execution.checkpointing.checkpoints-after-tasks-finish.enabled") .booleanType() diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/CheckpointConfig.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/CheckpointConfig.java index c404212d587..8308dcaee2c 100644 --- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/CheckpointConfig.java +++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/CheckpointConfig.java @@ -656,6 +656,16 @@ public class CheckpointConfig implements java.io.Serializable { return configuration.get(CheckpointingOptions.ENABLE_UNALIGNED); } + @Experimental + public void enableUnalignedCheckpointsInterruptibleTimers(boolean enabled) { + configuration.set(CheckpointingOptions.ENABLE_UNALIGNED_INTERRUPTIBLE_TIMERS, enabled); + } + + @Experimental + public boolean isUnalignedCheckpointsInterruptibleTimersEnabled() { + return configuration.get(CheckpointingOptions.ENABLE_UNALIGNED_INTERRUPTIBLE_TIMERS); + } + /** * Only relevant if {@link #isUnalignedCheckpointsEnabled} is enabled. * @@ -1065,6 +1075,9 @@ public class CheckpointConfig implements java.io.Serializable { configuration .getOptional(CheckpointingOptions.ENABLE_UNALIGNED) .ifPresent(this::enableUnalignedCheckpoints); + configuration + .getOptional(CheckpointingOptions.ENABLE_UNALIGNED_INTERRUPTIBLE_TIMERS) + .ifPresent(this::enableUnalignedCheckpointsInterruptibleTimers); configuration .getOptional(StateRecoveryOptions.CHECKPOINT_ID_OF_IGNORED_IN_FLIGHT_DATA) .ifPresent(this::setCheckpointIdOfIgnoredInFlightData); diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamConfig.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamConfig.java index eb98755d6b7..e56ffdbf10a 100644 --- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamConfig.java +++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamConfig.java @@ -532,6 +532,14 @@ public class StreamConfig implements Serializable { return config.get(CheckpointingOptions.ENABLE_UNALIGNED, false); } + public void setUnalignedCheckpointsSplittableTimersEnabled(boolean enabled) { + config.setBoolean(CheckpointingOptions.ENABLE_UNALIGNED_INTERRUPTIBLE_TIMERS, enabled); + } + + public boolean isUnalignedCheckpointsSplittableTimersEnabled() { + return config.get(CheckpointingOptions.ENABLE_UNALIGNED_INTERRUPTIBLE_TIMERS); + } + public boolean isExactlyOnceCheckpointMode() { return getCheckpointMode() == CheckpointingMode.EXACTLY_ONCE; } diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java index c8099a47913..cbe6513db5e 100644 --- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java +++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java @@ -1142,6 +1142,8 @@ public class StreamingJobGraphGenerator { streamGraph.isEnableCheckpointsAfterTasksFinish()); config.setCheckpointMode(getCheckpointingMode(checkpointCfg)); config.setUnalignedCheckpointsEnabled(checkpointCfg.isUnalignedCheckpointsEnabled()); + config.setUnalignedCheckpointsSplittableTimersEnabled( + checkpointCfg.isUnalignedCheckpointsInterruptibleTimersEnabled()); config.setAlignedCheckpointTimeout(checkpointCfg.getAlignedCheckpointTimeout()); config.setMaxSubtasksPerChannelStateFile(checkpointCfg.getMaxSubtasksPerChannelStateFile()); config.setMaxConcurrentCheckpoints(checkpointCfg.getMaxConcurrentCheckpoints()); diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java index ca82984af6a..e685d1e2b81 100644 --- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java +++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java @@ -24,6 +24,7 @@ import org.apache.flink.annotation.PublicEvolving; import org.apache.flink.annotation.VisibleForTesting; import org.apache.flink.api.common.ExecutionConfig; import org.apache.flink.api.common.eventtime.IndexedCombinedWatermarkStatus; +import org.apache.flink.api.common.operators.MailboxExecutor; import org.apache.flink.api.common.state.KeyedStateStore; import org.apache.flink.api.common.state.State; import org.apache.flink.api.common.state.StateDescriptor; @@ -63,6 +64,8 @@ import org.apache.flink.util.Preconditions; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import javax.annotation.Nullable; + import java.io.Serializable; import java.util.Arrays; import java.util.Collections; @@ -93,6 +96,7 @@ import static org.apache.flink.util.Preconditions.checkState; public abstract class AbstractStreamOperator<OUT> implements StreamOperator<OUT>, SetupableStreamOperator<OUT>, + YieldingOperator<OUT>, CheckpointedStreamOperator, KeyContextHandler, Serializable { @@ -120,6 +124,10 @@ public abstract class AbstractStreamOperator<OUT> /** The runtime context for UDFs. */ private transient StreamingRuntimeContext runtimeContext; + private transient @Nullable MailboxExecutor mailboxExecutor; + + private transient @Nullable MailboxWatermarkProcessor watermarkProcessor; + // ---------------- key/value state ------------------ /** @@ -311,6 +319,37 @@ public abstract class AbstractStreamOperator<OUT> return false; } + @Internal + @Override + public void setMailboxExecutor(MailboxExecutor mailboxExecutor) { + this.mailboxExecutor = mailboxExecutor; + } + + /** + * Can be overridden to disable splittable timers for this particular operator even if config + * option is enabled. By default, splittable timers are disabled. + * + * @return {@code true} if splittable timers should be used (subject to {@link + * StreamConfig#isUnalignedCheckpointsEnabled()} and {@link + * StreamConfig#isUnalignedCheckpointsSplittableTimersEnabled()}. {@code false} if + * splittable timers should never be used. + */ + @Internal + public boolean useSplittableTimers() { + return false; + } + + @Internal + private boolean areSplittableTimersConfigured() { + return areSplittableTimersConfigured(config); + } + + static boolean areSplittableTimersConfigured(StreamConfig config) { + return config.isCheckpointingEnabled() + && config.isUnalignedCheckpointsEnabled() + && config.isUnalignedCheckpointsSplittableTimersEnabled(); + } + /** * This method is called immediately before any elements are processed, it should contain the * operator's initialization logic, e.g. state initialization. @@ -320,7 +359,15 @@ public abstract class AbstractStreamOperator<OUT> * @throws Exception An exception in this method causes the operator to fail. */ @Override - public void open() throws Exception {} + public void open() throws Exception { + if (useSplittableTimers() + && areSplittableTimersConfigured() + && getTimeServiceManager().isPresent()) { + this.watermarkProcessor = + new MailboxWatermarkProcessor( + output, mailboxExecutor, getTimeServiceManager().get()); + } + } @Override public void finish() throws Exception {} @@ -618,6 +665,14 @@ public abstract class AbstractStreamOperator<OUT> } public void processWatermark(Watermark mark) throws Exception { + if (watermarkProcessor != null) { + watermarkProcessor.emitWatermarkInsideMailbox(mark); + } else { + emitWatermarkDirectly(mark); + } + } + + private void emitWatermarkDirectly(Watermark mark) throws Exception { if (timeServiceManager != null) { timeServiceManager.advanceWatermark(mark); } diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InternalTimeServiceManager.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InternalTimeServiceManager.java index f078401c045..a1b89031f5e 100644 --- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InternalTimeServiceManager.java +++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InternalTimeServiceManager.java @@ -40,6 +40,16 @@ import java.io.Serializable; */ @Internal public interface InternalTimeServiceManager<K> { + + /** Signals whether the watermark should continue advancing. */ + @Internal + @FunctionalInterface + interface ShouldStopAdvancingFn { + + /** @return {@code true} if firing timers should be interrupted. */ + boolean test(); + } + /** * Creates an {@link InternalTimerService} for handling a group of timers identified by the * given {@code name}. The timers are scoped to a key and namespace. @@ -73,6 +83,15 @@ public interface InternalTimeServiceManager<K> { */ void advanceWatermark(Watermark watermark) throws Exception; + /** + * Try to {@link #advanceWatermark(Watermark)}, but if {@link ShouldStopAdvancingFn} returns + * {@code true}, stop the advancement and return as soon as possible. + * + * @return true if {@link Watermark} has been fully processed, false otherwise. + */ + boolean tryAdvanceWatermark(Watermark watermark, ShouldStopAdvancingFn shouldStopAdvancingFn) + throws Exception; + /** * Snapshots the timers to raw keyed state. * diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InternalTimeServiceManagerImpl.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InternalTimeServiceManagerImpl.java index 886801c35ce..6bdfe05f4b9 100644 --- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InternalTimeServiceManagerImpl.java +++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InternalTimeServiceManagerImpl.java @@ -245,6 +245,17 @@ public class InternalTimeServiceManagerImpl<K> implements InternalTimeServiceMan } } + @Override + public boolean tryAdvanceWatermark( + Watermark watermark, ShouldStopAdvancingFn shouldStopAdvancingFn) throws Exception { + for (InternalTimerServiceImpl<?, ?> service : timerServices.values()) { + if (!service.tryAdvanceWatermark(watermark.getTimestamp(), shouldStopAdvancingFn)) { + return false; + } + } + return true; + } + ////////////////// Fault Tolerance Methods /////////////////// @Override diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InternalTimerServiceImpl.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InternalTimerServiceImpl.java index fc57c625613..269ba493b18 100644 --- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InternalTimerServiceImpl.java +++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InternalTimerServiceImpl.java @@ -307,18 +307,38 @@ public class InternalTimerServiceImpl<K, N> implements InternalTimerService<N> { } public void advanceWatermark(long time) throws Exception { - currentWatermark = time; + Preconditions.checkState( + tryAdvanceWatermark( + time, + () -> { + // Never stop advancing. + return false; + })); + } + /** + * @return true if following watermarks can be processed immediately. False if the firing timers + * should be interrupted as soon as possible. + */ + public boolean tryAdvanceWatermark( + long time, InternalTimeServiceManager.ShouldStopAdvancingFn shouldStopAdvancingFn) + throws Exception { + currentWatermark = time; InternalTimer<K, N> timer; - + boolean interrupted = false; while ((timer = eventTimeTimersQueue.peek()) != null && timer.getTimestamp() <= time - && !cancellationContext.isCancelled()) { + && !cancellationContext.isCancelled() + && !interrupted) { keyContext.setCurrentKey(timer.getKey()); eventTimeTimersQueue.poll(); triggerTarget.onEventTime(timer); taskIOMetricGroup.getNumFiredTimers().inc(); + // Check if we should stop advancing after at least one iteration to guarantee progress + // and prevent a potential starvation. + interrupted = shouldStopAdvancingFn.test(); } + return !interrupted; } /** diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/MailboxWatermarkProcessor.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/MailboxWatermarkProcessor.java new file mode 100644 index 00000000000..02cfd966c1c --- /dev/null +++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/MailboxWatermarkProcessor.java @@ -0,0 +1,92 @@ +/* + * 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.flink.streaming.api.operators; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.api.common.operators.MailboxExecutor; +import org.apache.flink.streaming.api.watermark.Watermark; +import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import static org.apache.flink.util.Preconditions.checkNotNull; + +/** + * A helper class to let operators emit watermarks incrementally from mailbox. Instead of emitting + * all the watermarks at once in a single {@code processWatermark} call, if a mail in mailbox is + * present, the process of firing timers is interrupted and a continuation to finish it off later is + * scheduled via a mailbox mail. + * + * <p>Note that interrupting firing timers can change order of some invocations. It is possible that + * between firing timers, some records might be processed. + */ +@Internal +public class MailboxWatermarkProcessor<OUT> { + protected static final Logger LOG = LoggerFactory.getLogger(MailboxWatermarkProcessor.class); + + private final Output<StreamRecord<OUT>> output; + private final MailboxExecutor mailboxExecutor; + private final InternalTimeServiceManager<?> internalTimeServiceManager; + /** + * Flag to indicate whether a progress watermark is scheduled in the mailbox. This is used to + * avoid duplicate scheduling in case we have multiple watermarks to process. + */ + private boolean progressWatermarkScheduled = false; + + private Watermark maxInputWatermark = Watermark.UNINITIALIZED; + + public MailboxWatermarkProcessor( + Output<StreamRecord<OUT>> output, + MailboxExecutor mailboxExecutor, + InternalTimeServiceManager<?> internalTimeServiceManager) { + this.output = checkNotNull(output); + this.mailboxExecutor = checkNotNull(mailboxExecutor); + this.internalTimeServiceManager = checkNotNull(internalTimeServiceManager); + } + + public void emitWatermarkInsideMailbox(Watermark mark) throws Exception { + maxInputWatermark = + new Watermark(Math.max(maxInputWatermark.getTimestamp(), mark.getTimestamp())); + emitWatermarkInsideMailbox(); + } + + private void emitWatermarkInsideMailbox() throws Exception { + // Try to progress min watermark as far as we can. + if (internalTimeServiceManager.tryAdvanceWatermark( + maxInputWatermark, mailboxExecutor::shouldInterrupt)) { + // In case output watermark has fully progressed emit it downstream. + output.emitWatermark(maxInputWatermark); + } else if (!progressWatermarkScheduled) { + progressWatermarkScheduled = true; + // We still have work to do, but we need to let other mails to be processed first. + mailboxExecutor.execute( + () -> { + progressWatermarkScheduled = false; + emitWatermarkInsideMailbox(); + }, + "emitWatermarkInsideMailbox"); + } else { + // We're not guaranteed that MailboxProcessor is going to process all mails before + // processing additional input, so the advanceWatermark could be called before the + // previous watermark is fully processed. + LOG.debug("emitWatermarkInsideMailbox is already scheduled, skipping."); + } + } +} diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamOperatorFactoryUtil.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamOperatorFactoryUtil.java index 6f6ff1c2de2..016a7d8b302 100644 --- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamOperatorFactoryUtil.java +++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamOperatorFactoryUtil.java @@ -88,7 +88,7 @@ public class StreamOperatorFactoryUtil { : processingTimeServiceFactory, operatorEventDispatcher)); if (op instanceof YieldingOperator) { - ((YieldingOperator<?>) operatorFactory).setMailboxExecutor(mailboxExecutor); + ((YieldingOperator<?>) op).setMailboxExecutor(mailboxExecutor); } return new Tuple2<>(op, Optional.ofNullable(processingTimeService)); } diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionInternalTimeServiceManager.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionInternalTimeServiceManager.java index 32da4c5b4ae..bc08bf635bf 100644 --- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionInternalTimeServiceManager.java +++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionInternalTimeServiceManager.java @@ -91,6 +91,13 @@ public class BatchExecutionInternalTimeServiceManager<K> } } + @Override + public boolean tryAdvanceWatermark( + Watermark watermark, ShouldStopAdvancingFn shouldStopAdvancingFn) { + advanceWatermark(watermark); + return true; + } + @Override public void snapshotToRawKeyedState( KeyedStateCheckpointOutputStream context, String operatorName) throws Exception { diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java index 6c70dc47039..e830b8a0343 100644 --- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java +++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java @@ -954,6 +954,11 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>> } } + @VisibleForTesting + public boolean runSingleMailboxLoop() throws Exception { + return mailboxProcessor.runSingleMailboxLoop(); + } + @VisibleForTesting public boolean runMailboxStep() throws Exception { return mailboxProcessor.runMailboxStep(); @@ -1126,6 +1131,10 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>> return this.mailboxProcessor::getMailboxExecutor; } + public boolean hasMail() { + return mailboxProcessor.hasMail(); + } + private boolean taskIsAvailable() { return recordWriter.isAvailable() && (changelogWriterAvailabilityProvider == null diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxProcessor.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxProcessor.java index 3dae1cec3df..1a6f0694ed8 100644 --- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxProcessor.java +++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxProcessor.java @@ -239,6 +239,22 @@ public class MailboxProcessor implements Closeable { sendPoisonMail(() -> suspended = true); } + /** + * Execute a single (as small as possible) step of the mailbox. + * + * @return true if something was processed. + */ + @VisibleForTesting + public boolean runSingleMailboxLoop() throws Exception { + suspended = !mailboxLoopRunning; + boolean processed = processMail(mailbox, true); + if (isDefaultActionAvailable() && isNextLoopPossible()) { + mailboxDefaultAction.runDefaultAction(new MailboxController(this)); + processed = true; + } + return processed; + } + /** * Execute a single (as small as possible) step of the mailbox. * diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/environment/CheckpointConfigFromConfigurationTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/environment/CheckpointConfigFromConfigurationTest.java index 40c4131204e..d97ca4b9c30 100644 --- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/environment/CheckpointConfigFromConfigurationTest.java +++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/environment/CheckpointConfigFromConfigurationTest.java @@ -151,6 +151,14 @@ public class CheckpointConfigFromConfigurationTest { .viaSetter(CheckpointConfig::enableUnalignedCheckpoints) .getterVia(CheckpointConfig::isUnalignedCheckpointsEnabled) .nonDefaultValue(true), + TestSpec.testValue(true) + .whenSetFromFile( + "execution.checkpointing.unaligned.interruptible-timers.enabled", + "true") + .viaSetter(CheckpointConfig::enableUnalignedCheckpointsInterruptibleTimers) + .getterVia( + CheckpointConfig::isUnalignedCheckpointsInterruptibleTimersEnabled) + .nonDefaultValue(true), TestSpec.testValue( (CheckpointStorage) new FileSystemCheckpointStorage( diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/MailboxWatermarkProcessorTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/MailboxWatermarkProcessorTest.java new file mode 100644 index 00000000000..9fb9d348525 --- /dev/null +++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/MailboxWatermarkProcessorTest.java @@ -0,0 +1,120 @@ +/* + * 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.flink.streaming.api.operators; + +import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.runtime.asyncprocessing.AsyncExecutionController; +import org.apache.flink.runtime.state.KeyedStateCheckpointOutputStream; +import org.apache.flink.streaming.api.watermark.Watermark; +import org.apache.flink.streaming.runtime.streamrecord.StreamElement; +import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; +import org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor; +import org.apache.flink.streaming.runtime.tasks.mailbox.Mail; +import org.apache.flink.streaming.runtime.tasks.mailbox.MailboxExecutorImpl; +import org.apache.flink.streaming.runtime.tasks.mailbox.TaskMailbox; +import org.apache.flink.streaming.runtime.tasks.mailbox.TaskMailboxImpl; +import org.apache.flink.streaming.util.CollectorOutput; + +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for {@link MailboxWatermarkProcessor}. */ +class MailboxWatermarkProcessorTest { + + @Test + void testEmitWatermarkInsideMailbox() throws Exception { + final List<StreamElement> emittedElements = new ArrayList<>(); + final TaskMailboxImpl mailbox = new TaskMailboxImpl(); + final InternalTimeServiceManager<?> timerService = new NoOpInternalTimeServiceManager(); + + final MailboxWatermarkProcessor<StreamRecord<String>> watermarkProcessor = + new MailboxWatermarkProcessor<>( + new CollectorOutput<>(emittedElements), + new MailboxExecutorImpl(mailbox, 0, StreamTaskActionExecutor.IMMEDIATE), + timerService); + final List<Watermark> expectedOutput = new ArrayList<>(); + watermarkProcessor.emitWatermarkInsideMailbox(new Watermark(1)); + watermarkProcessor.emitWatermarkInsideMailbox(new Watermark(2)); + watermarkProcessor.emitWatermarkInsideMailbox(new Watermark(3)); + expectedOutput.add(new Watermark(1)); + expectedOutput.add(new Watermark(2)); + expectedOutput.add(new Watermark(3)); + + assertThat(emittedElements).containsExactlyElementsOf(expectedOutput); + + mailbox.put(new Mail(() -> {}, TaskMailbox.MIN_PRIORITY, "checkpoint mail")); + + watermarkProcessor.emitWatermarkInsideMailbox(new Watermark(4)); + watermarkProcessor.emitWatermarkInsideMailbox(new Watermark(5)); + + assertThat(emittedElements).containsExactlyElementsOf(expectedOutput); + + while (mailbox.hasMail()) { + mailbox.take(TaskMailbox.MIN_PRIORITY).run(); + } + // Watermark(4) is processed together with Watermark(5) + expectedOutput.add(new Watermark(5)); + + assertThat(emittedElements).containsExactlyElementsOf(expectedOutput); + } + + private static class NoOpInternalTimeServiceManager + implements InternalTimeServiceManager<Object> { + @Override + public <N> InternalTimerService<N> getInternalTimerService( + String name, + TypeSerializer<Object> keySerializer, + TypeSerializer<N> namespaceSerializer, + Triggerable<Object, N> triggerable) { + throw new UnsupportedOperationException(); + } + + @Override + public <N> InternalTimerService<N> getAsyncInternalTimerService( + String name, + TypeSerializer<Object> keySerializer, + TypeSerializer<N> namespaceSerializer, + Triggerable<Object, N> triggerable, + AsyncExecutionController<Object> asyncExecutionController) { + throw new UnsupportedOperationException(); + } + + @Override + public void advanceWatermark(Watermark watermark) throws Exception { + throw new UnsupportedOperationException(); + } + + @Override + public boolean tryAdvanceWatermark( + Watermark watermark, ShouldStopAdvancingFn shouldStopAdvancingFn) throws Exception { + return !shouldStopAdvancingFn.test(); + } + + @Override + public void snapshotToRawKeyedState( + KeyedStateCheckpointOutputStream stateCheckpointOutputStream, String operatorName) + throws Exception { + throw new UnsupportedOperationException(); + } + } +} diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/UnalignedCheckpointsInterruptibleTimersTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/UnalignedCheckpointsInterruptibleTimersTest.java new file mode 100644 index 00000000000..a5e3712ec76 --- /dev/null +++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/UnalignedCheckpointsInterruptibleTimersTest.java @@ -0,0 +1,230 @@ +/* + * 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.flink.streaming.runtime.io.checkpointing; + +import org.apache.flink.api.common.operators.MailboxExecutor; +import org.apache.flink.api.common.typeinfo.Types; +import org.apache.flink.api.common.typeutils.base.StringSerializer; +import org.apache.flink.streaming.api.graph.StreamConfig; +import org.apache.flink.streaming.api.operators.AbstractStreamOperator; +import org.apache.flink.streaming.api.operators.InternalTimer; +import org.apache.flink.streaming.api.operators.InternalTimerService; +import org.apache.flink.streaming.api.operators.MailboxWatermarkProcessor; +import org.apache.flink.streaming.api.operators.OneInputStreamOperator; +import org.apache.flink.streaming.api.operators.SimpleOperatorFactory; +import org.apache.flink.streaming.api.operators.Triggerable; +import org.apache.flink.streaming.api.operators.YieldingOperator; +import org.apache.flink.streaming.api.watermark.Watermark; +import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; +import org.apache.flink.streaming.runtime.tasks.OneInputStreamTask; +import org.apache.flink.streaming.runtime.tasks.StreamTaskMailboxTestHarness; +import org.apache.flink.streaming.runtime.tasks.StreamTaskMailboxTestHarnessBuilder; +import org.apache.flink.util.TestLoggerExtension; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; + +import javax.annotation.Nullable; + +import java.time.Instant; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Tests interaction between {@link CheckpointingOptions#ENABLE_UNALIGNED_INTERRUPTIBLE_TIMERS} and + * unaligned checkpoints. + */ +@ExtendWith(TestLoggerExtension.class) +class UnalignedCheckpointsInterruptibleTimersTest { + + @Test + void testSingleWatermarkHoldingOperatorInTheChain() throws Exception { + final Instant firstWindowEnd = Instant.ofEpochMilli(1000L); + final int numFirstWindowTimers = 2; + final Instant secondWindowEnd = Instant.ofEpochMilli(2000L); + final int numSecondWindowTimers = 2; + + try (final StreamTaskMailboxTestHarness<String> harness = + new StreamTaskMailboxTestHarnessBuilder<>(OneInputStreamTask::new, Types.STRING) + .modifyStreamConfig( + UnalignedCheckpointsInterruptibleTimersTest::setupStreamConfig) + .addInput(Types.STRING) + .setupOperatorChain( + SimpleOperatorFactory.of( + new MultipleTimersAtTheSameTimestamp() + .withTimers(firstWindowEnd, numFirstWindowTimers) + .withTimers( + secondWindowEnd, numSecondWindowTimers))) + .name("first") + .finishForSingletonOperatorChain(StringSerializer.INSTANCE) + .build()) { + harness.processElement(new StreamRecord<>("register timers")); + harness.processAll(); + harness.processElement(asWatermark(firstWindowEnd)); + harness.processElement(asWatermark(secondWindowEnd)); + + assertThat(harness.getOutput()) + .containsExactly( + asFiredRecord("key-0"), + asMailRecord("key-0"), + asFiredRecord("key-1"), + asMailRecord("key-1"), + asWatermark(firstWindowEnd), + asFiredRecord("key-0"), + asMailRecord("key-0"), + asFiredRecord("key-1"), + asMailRecord("key-1"), + asWatermark(secondWindowEnd)); + } + } + + @Test + void testWatermarkProgressWithNoTimers() throws Exception { + final Instant firstWindowEnd = Instant.ofEpochMilli(1000L); + final Instant secondWindowEnd = Instant.ofEpochMilli(2000L); + + try (final StreamTaskMailboxTestHarness<String> harness = + new StreamTaskMailboxTestHarnessBuilder<>(OneInputStreamTask::new, Types.STRING) + .modifyStreamConfig( + UnalignedCheckpointsInterruptibleTimersTest::setupStreamConfig) + .addInput(Types.STRING) + .setupOperatorChain( + SimpleOperatorFactory.of(new MultipleTimersAtTheSameTimestamp())) + .name("first") + .finishForSingletonOperatorChain(StringSerializer.INSTANCE) + .build()) { + harness.setAutoProcess(false); + harness.processElement(new StreamRecord<>("impulse")); + harness.processAll(); + harness.processElement(asWatermark(firstWindowEnd)); + harness.processElement(asWatermark(secondWindowEnd)); + + final List<Watermark> seenWatermarks = new ArrayList<>(); + while (seenWatermarks.size() < 2) { + harness.processSingleStep(); + Object outputElement; + while ((outputElement = harness.getOutput().poll()) != null) { + if (outputElement instanceof Watermark) { + seenWatermarks.add((Watermark) outputElement); + } + } + } + assertThat(seenWatermarks) + .containsExactly(asWatermark(firstWindowEnd), asWatermark(secondWindowEnd)); + } + } + + private static Watermark asWatermark(Instant timestamp) { + return new Watermark(timestamp.toEpochMilli()); + } + + private static StreamRecord<String> asFiredRecord(String key) { + return new StreamRecord("fired-" + key); + } + + private static StreamRecord<String> asMailRecord(String key) { + return new StreamRecord("mail-" + key); + } + + private static void setupStreamConfig(StreamConfig cfg) { + cfg.setUnalignedCheckpointsEnabled(true); + cfg.setUnalignedCheckpointsSplittableTimersEnabled(true); + cfg.setStateKeySerializer(StringSerializer.INSTANCE); + } + + private static class MultipleTimersAtTheSameTimestamp extends AbstractStreamOperator<String> + implements OneInputStreamOperator<String, String>, + Triggerable<String, String>, + YieldingOperator<String> { + + private final Map<Instant, Integer> timersToRegister; + private transient @Nullable MailboxExecutor mailboxExecutor; + private transient @Nullable MailboxWatermarkProcessor watermarkProcessor; + + MultipleTimersAtTheSameTimestamp() { + this(Collections.emptyMap()); + } + + MultipleTimersAtTheSameTimestamp(Map<Instant, Integer> timersToRegister) { + this.timersToRegister = timersToRegister; + } + + @Override + public void setMailboxExecutor(MailboxExecutor mailboxExecutor) { + this.mailboxExecutor = mailboxExecutor; + } + + @Override + public void open() throws Exception { + super.open(); + if (getTimeServiceManager().isPresent()) { + this.watermarkProcessor = + new MailboxWatermarkProcessor( + output, mailboxExecutor, getTimeServiceManager().get()); + } + } + + @Override + public void processElement(StreamRecord<String> element) { + if (!timersToRegister.isEmpty()) { + final InternalTimerService<String> timers = + getInternalTimerService("timers", StringSerializer.INSTANCE, this); + for (Map.Entry<Instant, Integer> entry : timersToRegister.entrySet()) { + for (int keyIdx = 0; keyIdx < entry.getValue(); keyIdx++) { + final String key = String.format("key-%d", keyIdx); + setCurrentKey(key); + timers.registerEventTimeTimer( + String.format("window-%s", entry.getKey()), + entry.getKey().toEpochMilli()); + } + } + } + } + + @Override + public void processWatermark(Watermark mark) throws Exception { + if (watermarkProcessor == null) { + super.processWatermark(mark); + } else { + watermarkProcessor.emitWatermarkInsideMailbox(mark); + } + } + + @Override + public void onEventTime(InternalTimer<String, String> timer) throws Exception { + mailboxExecutor.execute( + () -> output.collect(asMailRecord(timer.getKey())), "mail-" + timer.getKey()); + output.collect(asFiredRecord(timer.getKey())); + } + + @Override + public void onProcessingTime(InternalTimer<String, String> timer) throws Exception {} + + MultipleTimersAtTheSameTimestamp withTimers(Instant timestamp, int count) { + final Map<Instant, Integer> copy = new HashMap<>(timersToRegister); + copy.put(timestamp, count); + return new MultipleTimersAtTheSameTimestamp(copy); + } + } +} diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/UnalignedCheckpointsTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/UnalignedCheckpointsTest.java index 82c06c851f8..53e59680a7d 100644 --- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/UnalignedCheckpointsTest.java +++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/checkpointing/UnalignedCheckpointsTest.java @@ -548,7 +548,7 @@ class UnalignedCheckpointsTest { } @Test - void testNotifyAbortCheckpointBeforeCanellingAsyncCheckpoint() throws Exception { + public void testNotifyAbortCheckpointBeforeCancellingAsyncCheckpoint() throws Exception { ValidateAsyncFutureNotCompleted handler = new ValidateAsyncFutureNotCompleted(1); inputGate = createInputGate(2, handler); handler.setInputGate(inputGate); @@ -699,7 +699,7 @@ class UnalignedCheckpointsTest { * Tests {@link * SingleCheckpointBarrierHandler#processCancellationBarrier(CancelCheckpointMarker, * InputChannelInfo)} abort the current pending checkpoint triggered by {@link - * CheckpointBarrierHandler#processBarrier(CheckpointBarrier, InputChannelInfo)}. + * CheckpointBarrierHandler#processBarrier(CheckpointBarrier, InputChannelInfo, boolean)}. */ @Test void testProcessCancellationBarrierAfterProcessBarrier() throws Exception { diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/TableStreamOperator.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/TableStreamOperator.java index b8717bb1ce9..d91a9b3102e 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/TableStreamOperator.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/TableStreamOperator.java @@ -52,12 +52,6 @@ public abstract class TableStreamOperator<OUT> extends AbstractStreamOperator<OU this.ctx = new ContextImpl(getProcessingTimeService()); } - @Override - public void processWatermark(Watermark mark) throws Exception { - super.processWatermark(mark); - currentWatermark = mark.getTimestamp(); - } - /** Compute memory size from memory faction. */ public long computeMemorySize() { final Environment environment = getContainingTask().getEnvironment(); @@ -72,6 +66,11 @@ public abstract class TableStreamOperator<OUT> extends AbstractStreamOperator<OU environment.getUserCodeClassLoader().asClassLoader())); } + @Override + public void processWatermark(Watermark mark) throws Exception { + currentWatermark = mark.getTimestamp(); + super.processWatermark(mark); + } /** Information available in an invocation of processElement. */ protected class ContextImpl implements TimerService { diff --git a/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/streaming/util/TestStreamEnvironment.java b/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/streaming/util/TestStreamEnvironment.java index f233f80c36c..ec595609437 100644 --- a/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/streaming/util/TestStreamEnvironment.java +++ b/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/streaming/util/TestStreamEnvironment.java @@ -120,6 +120,8 @@ public class TestStreamEnvironment extends StreamExecutionEnvironment { Duration.ofMillis(100), Duration.ofSeconds(2)); randomize(conf, CheckpointingOptions.CLEANER_PARALLEL_MODE, true, false); + randomize( + conf, CheckpointingOptions.ENABLE_UNALIGNED_INTERRUPTIBLE_TIMERS, true, false); randomize(conf, ExecutionOptions.SNAPSHOT_COMPRESSION, true, false); if (!conf.contains(CheckpointingOptions.FILE_MERGING_ENABLED)) { randomize(conf, CheckpointingOptions.FILE_MERGING_ENABLED, true);