dawidwys commented on a change in pull request #13443:
URL: https://github.com/apache/flink/pull/13443#discussion_r493754193
##########
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InternalTimeServiceManager.java
##########
@@ -19,214 +19,62 @@
package org.apache.flink.streaming.api.operators;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.core.memory.DataOutputView;
-import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
-import org.apache.flink.runtime.state.KeyGroupRange;
-import org.apache.flink.runtime.state.KeyGroupedInternalPriorityQueue;
-import org.apache.flink.runtime.state.KeyGroupsList;
-import org.apache.flink.runtime.state.KeyedStateBackend;
-import org.apache.flink.runtime.state.KeyedStateCheckpointOutputStream;
-import org.apache.flink.runtime.state.PriorityQueueSetFactory;
+import org.apache.flink.runtime.state.CheckpointableKeyedStateBackend;
+import org.apache.flink.runtime.state.KeyGroupStatePartitionStreamProvider;
import org.apache.flink.runtime.state.StateSnapshotContext;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
-import org.apache.flink.util.Preconditions;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
-
-import static org.apache.flink.util.Preconditions.checkNotNull;
/**
* An entity keeping all the time-related services available to all operators
extending the
- * {@link AbstractStreamOperator}. Right now, this is only a
- * {@link InternalTimerServiceImpl timer services}.
+ * {@link AbstractStreamOperator} or {@link AbstractStreamOperatorV2}.
*
* <b>NOTE:</b> These services are only available to keyed operators.
*
* @param <K> The type of keys used for the timers and the registry.
*/
@Internal
-public class InternalTimeServiceManager<K> {
- protected static final Logger LOG =
LoggerFactory.getLogger(InternalTimeServiceManager.class);
-
- @VisibleForTesting
- static final String TIMER_STATE_PREFIX = "_timer_state";
- @VisibleForTesting
- static final String PROCESSING_TIMER_PREFIX = TIMER_STATE_PREFIX +
"/processing_";
- @VisibleForTesting
- static final String EVENT_TIMER_PREFIX = TIMER_STATE_PREFIX + "/event_";
-
- private final KeyGroupRange localKeyGroupRange;
- private final KeyContext keyContext;
-
- private final PriorityQueueSetFactory priorityQueueSetFactory;
- private final ProcessingTimeService processingTimeService;
-
- private final Map<String, InternalTimerServiceImpl<K, ?>> timerServices;
-
- private final boolean useLegacySynchronousSnapshots;
-
- InternalTimeServiceManager(
- KeyGroupRange localKeyGroupRange,
- KeyContext keyContext,
- PriorityQueueSetFactory priorityQueueSetFactory,
- ProcessingTimeService processingTimeService,
- boolean useLegacySynchronousSnapshots) {
-
- this.localKeyGroupRange =
Preconditions.checkNotNull(localKeyGroupRange);
- this.priorityQueueSetFactory =
Preconditions.checkNotNull(priorityQueueSetFactory);
- this.keyContext = Preconditions.checkNotNull(keyContext);
- this.processingTimeService =
Preconditions.checkNotNull(processingTimeService);
- this.useLegacySynchronousSnapshots =
useLegacySynchronousSnapshots;
-
- this.timerServices = new HashMap<>();
- }
-
- public <N> InternalTimerService<N> getInternalTimerService(
- String name,
- TypeSerializer<N> namespaceSerializer,
- Triggerable<K, N> triggerable,
- KeyedStateBackend<K> keyedStateBackend) {
- checkNotNull(keyedStateBackend, "Timers can only be used on
keyed operators.");
-
- TypeSerializer<K> keySerializer =
keyedStateBackend.getKeySerializer();
- // the following casting is to overcome type restrictions.
- TimerSerializer<K, N> timerSerializer = new
TimerSerializer<>(keySerializer, namespaceSerializer);
- return getInternalTimerService(name, timerSerializer,
triggerable);
- }
-
- public <N> InternalTimerService<N> getInternalTimerService(
+public interface InternalTimeServiceManager<K> {
+ /**
+ * 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.
+ *
+ * <p>When the timer is due it will call the given {@link Triggerable}.
+ */
+ <N> InternalTimerService<N> getInternalTimerService(
String name,
- TimerSerializer<K, N> timerSerializer,
- Triggerable<K, N> triggerable) {
-
- InternalTimerServiceImpl<K, N> timerService =
registerOrGetTimerService(name, timerSerializer);
-
- timerService.startTimerService(
- timerSerializer.getKeySerializer(),
- timerSerializer.getNamespaceSerializer(),
- triggerable);
-
- return timerService;
- }
-
- @SuppressWarnings("unchecked")
- <N> InternalTimerServiceImpl<K, N> registerOrGetTimerService(String
name, TimerSerializer<K, N> timerSerializer) {
- InternalTimerServiceImpl<K, N> timerService =
(InternalTimerServiceImpl<K, N>) timerServices.get(name);
- if (timerService == null) {
-
- timerService = new InternalTimerServiceImpl<>(
- localKeyGroupRange,
- keyContext,
- processingTimeService,
-
createTimerPriorityQueue(PROCESSING_TIMER_PREFIX + name, timerSerializer),
- createTimerPriorityQueue(EVENT_TIMER_PREFIX +
name, timerSerializer));
-
- timerServices.put(name, timerService);
- }
- return timerService;
- }
-
- Map<String, InternalTimerServiceImpl<K, ?>>
getRegisteredTimerServices() {
- return Collections.unmodifiableMap(timerServices);
- }
-
- private <N> KeyGroupedInternalPriorityQueue<TimerHeapInternalTimer<K,
N>> createTimerPriorityQueue(
- String name,
- TimerSerializer<K, N> timerSerializer) {
- return priorityQueueSetFactory.create(
- name,
- timerSerializer);
- }
-
- public void advanceWatermark(Watermark watermark) throws Exception {
- for (InternalTimerServiceImpl<?, ?> service :
timerServices.values()) {
- service.advanceWatermark(watermark.getTimestamp());
- }
- }
-
- ////////////////// Fault Tolerance Methods
///////////////////
-
- public void snapshotState(StateSnapshotContext context, String
operatorName) throws Exception {
- //TODO all of this can be removed once heap-based timers are
integrated with RocksDB incremental snapshots
- if (useLegacySynchronousSnapshots) {
- KeyedStateCheckpointOutputStream out;
- try {
- out = context.getRawKeyedOperatorStateOutput();
- } catch (Exception exception) {
- throw new Exception("Could not open raw keyed
operator state stream for " +
- operatorName + '.', exception);
- }
-
- try {
- KeyGroupsList allKeyGroups =
out.getKeyGroupList();
- for (int keyGroupIdx : allKeyGroups) {
- out.startNewKeyGroup(keyGroupIdx);
-
- snapshotStateForKeyGroup(
- new
DataOutputViewStreamWrapper(out), keyGroupIdx);
- }
- } catch (Exception exception) {
- throw new Exception("Could not write timer
service of " + operatorName +
- " to checkpoint state stream.",
exception);
- } finally {
- try {
- out.close();
- } catch (Exception closeException) {
- LOG.warn("Could not close raw keyed
operator state stream for {}. This " +
- "might have prevented deleting
some state data.", operatorName, closeException);
- }
- }
- }
- }
-
- private void snapshotStateForKeyGroup(DataOutputView stream, int
keyGroupIdx) throws IOException {
- InternalTimerServiceSerializationProxy<K> serializationProxy =
- new InternalTimerServiceSerializationProxy<>(this,
keyGroupIdx);
-
- serializationProxy.write(stream);
- }
-
- public void restoreStateForKeyGroup(
- InputStream stream,
- int keyGroupIdx,
- ClassLoader userCodeClassLoader) throws IOException {
-
- InternalTimerServiceSerializationProxy<K> serializationProxy =
- new InternalTimerServiceSerializationProxy<>(
- this,
- userCodeClassLoader,
- keyGroupIdx);
-
- serializationProxy.read(stream);
- }
-
- //////////////////// Methods used ONLY IN TESTS
////////////////////
-
- @VisibleForTesting
- public int numProcessingTimeTimers() {
- int count = 0;
- for (InternalTimerServiceImpl<?, ?> timerService :
timerServices.values()) {
- count += timerService.numProcessingTimeTimers();
- }
- return count;
- }
-
- @VisibleForTesting
- public int numEventTimeTimers() {
- int count = 0;
- for (InternalTimerServiceImpl<?, ?> timerService :
timerServices.values()) {
- count += timerService.numEventTimeTimers();
- }
- return count;
+ TypeSerializer<K> keySerializer,
+ TypeSerializer<N> namespaceSerializer,
+ Triggerable<K, N> triggerable);
+
+ /**
+ * It advances the Watermark of all managed {@link InternalTimerService
timer services},
+ * potentially firing the event time timers.
+ */
+ void advanceWatermark(Watermark watermark) throws Exception;
+
+ /**
+ * Snapshots the timers in the keyed state.
+ *
+ * <p><b>TODO:</b> This can be removed once heap-based timers are
integrated with RocksDB
+ * incremental snapshots.
+ */
+ void snapshotState(
+ StateSnapshotContext context,
+ String operatorName) throws Exception;
+
+ /**
+ * A provider pattern for creating an instance of a {@link
InternalTimeServiceManager}.
+ * Allows substituting the manager that will be used in the runtime.
+ */
+ @FunctionalInterface
+ interface Provider {
+ <K> InternalTimeServiceManager<K> create(
+ CheckpointableKeyedStateBackend<K> keyedStatedBackend,
+ ClassLoader userClassloader,
+ KeyContext keyContext, //the operator
Review comment:
actually it is (in the implementation) ;), but true it is misleading in
the interface
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
