@Test public void testKeyGroupStartIndexSetting() { int startKeyGroupIdx = 7; int endKeyGroupIdx = 21; KeyGroupRange testKeyGroupList = new KeyGroupRange(startKeyGroupIdx, endKeyGroupIdx); TestKeyContext keyContext = new TestKeyContext(); TestProcessingTimeService processingTimeService = new TestProcessingTimeService(); InternalTimerServiceImpl<Integer, String> service = createInternalTimerService( testKeyGroupList, keyContext, processingTimeService, IntSerializer.INSTANCE, StringSerializer.INSTANCE, createQueueFactory()); Assert.assertEquals(startKeyGroupIdx, service.getLocalKeyGroupRangeStartIdx()); }
@VisibleForTesting int getLocalKeyGroupRangeStartIdx() { return this.localKeyGroupRangeStartIdx; }
InternalTimerServiceImpl implements InternalTimerService<N>, ProcessingTimeCallback { @VisibleForTesting int getLocalKeyGroupRangeStartIdx() { return this.localKeyGroupRangeStartIdx; } }
InternalTimerServiceImpl implements InternalTimerService<N>, ProcessingTimeCallback { @VisibleForTesting int getLocalKeyGroupRangeStartIdx() { return this.localKeyGroupRangeStartIdx; } InternalTimerServiceImpl(
		KeyGroupRange localKeyGroupRange,
		KeyContext keyContext,
		ProcessingTimeService processingTimeService,
		KeyGroupedInternalPriorityQueue<TimerHeapInternalTimer<K, N>> processingTimeTimersQueue,
		KeyGroupedInternalPriorityQueue<TimerHeapInternalTimer<K, N>> eventTimeTimersQueue); }
InternalTimerServiceImpl implements InternalTimerService<N>, ProcessingTimeCallback { @VisibleForTesting int getLocalKeyGroupRangeStartIdx() { return this.localKeyGroupRangeStartIdx; } InternalTimerServiceImpl(
		KeyGroupRange localKeyGroupRange,
		KeyContext keyContext,
		ProcessingTimeService processingTimeService,
		KeyGroupedInternalPriorityQueue<TimerHeapInternalTimer<K, N>> processingTimeTimersQueue,
		KeyGroupedInternalPriorityQueue<TimerHeapInternalTimer<K, N>> eventTimeTimersQueue); void startTimerService(
			TypeSerializer<K> keySerializer,
			TypeSerializer<N> namespaceSerializer,
			Triggerable<K, N> triggerTarget); @Override long currentProcessingTime(); @Override long currentWatermark(); @Override void registerProcessingTimeTimer(N namespace, long time); @Override void registerEventTimeTimer(N namespace, long time); @Override void deleteProcessingTimeTimer(N namespace, long time); @Override void deleteEventTimeTimer(N namespace, long time); @Override void forEachEventTimeTimer(BiConsumerWithException<N, Long, Exception> consumer); @Override void forEachProcessingTimeTimer(BiConsumerWithException<N, Long, Exception> consumer); @Override void onProcessingTime(long time); void advanceWatermark(long time); InternalTimersSnapshot<K, N> snapshotTimersForKeyGroup(int keyGroupIdx); TypeSerializer<K> getKeySerializer(); TypeSerializer<N> getNamespaceSerializer(); @SuppressWarnings("unchecked") void restoreTimersForKeyGroup(InternalTimersSnapshot<?, ?> restoredSnapshot, int keyGroupIdx); @VisibleForTesting int numProcessingTimeTimers(); @VisibleForTesting int numEventTimeTimers(); @VisibleForTesting int numProcessingTimeTimers(N namespace); @VisibleForTesting int numEventTimeTimers(N namespace); }
InternalTimerServiceImpl implements InternalTimerService<N>, ProcessingTimeCallback { @VisibleForTesting int getLocalKeyGroupRangeStartIdx() { return this.localKeyGroupRangeStartIdx; } InternalTimerServiceImpl(
		KeyGroupRange localKeyGroupRange,
		KeyContext keyContext,
		ProcessingTimeService processingTimeService,
		KeyGroupedInternalPriorityQueue<TimerHeapInternalTimer<K, N>> processingTimeTimersQueue,
		KeyGroupedInternalPriorityQueue<TimerHeapInternalTimer<K, N>> eventTimeTimersQueue); void startTimerService(
			TypeSerializer<K> keySerializer,
			TypeSerializer<N> namespaceSerializer,
			Triggerable<K, N> triggerTarget); @Override long currentProcessingTime(); @Override long currentWatermark(); @Override void registerProcessingTimeTimer(N namespace, long time); @Override void registerEventTimeTimer(N namespace, long time); @Override void deleteProcessingTimeTimer(N namespace, long time); @Override void deleteEventTimeTimer(N namespace, long time); @Override void forEachEventTimeTimer(BiConsumerWithException<N, Long, Exception> consumer); @Override void forEachProcessingTimeTimer(BiConsumerWithException<N, Long, Exception> consumer); @Override void onProcessingTime(long time); void advanceWatermark(long time); InternalTimersSnapshot<K, N> snapshotTimersForKeyGroup(int keyGroupIdx); TypeSerializer<K> getKeySerializer(); TypeSerializer<N> getNamespaceSerializer(); @SuppressWarnings("unchecked") void restoreTimersForKeyGroup(InternalTimersSnapshot<?, ?> restoredSnapshot, int keyGroupIdx); @VisibleForTesting int numProcessingTimeTimers(); @VisibleForTesting int numEventTimeTimers(); @VisibleForTesting int numProcessingTimeTimers(N namespace); @VisibleForTesting int numEventTimeTimers(N namespace);  }