HeapInternalTimerService implements InternalTimerService<N>, ProcessingTimeCallback { @VisibleForTesting public int getLocalKeyGroupRangeStartIdx() { return this.localKeyGroupRangeStartIdx; } HeapInternalTimerService(
		int totalKeyGroups,
		KeyGroupsList localKeyGroupRange,
		KeyContext keyContext,
		ProcessingTimeService processingTimeService); 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 onProcessingTime(long time); void advanceWatermark(long time); InternalTimersSnapshot<K, N> snapshotTimersForKeyGroup(int keyGroupIdx); @SuppressWarnings("unchecked") void restoreTimersForKeyGroup(InternalTimersSnapshot<?, ?> restoredTimersSnapshot, int keyGroupIdx); int numProcessingTimeTimers(); int numEventTimeTimers(); int numProcessingTimeTimers(N namespace); int numEventTimeTimers(N namespace); @VisibleForTesting int getLocalKeyGroupRangeStartIdx(); @VisibleForTesting Set<InternalTimer<K, N>>[] getEventTimeTimersPerKeyGroup(); @VisibleForTesting Set<InternalTimer<K, N>>[] getProcessingTimeTimersPerKeyGroup();  }
@Test public void testKeyGroupStartIndexSetting() { int startKeyGroupIdx = 7; int endKeyGroupIdx = 21; KeyGroupsList testKeyGroupList = new KeyGroupRange(startKeyGroupIdx, endKeyGroupIdx); TestKeyContext keyContext = new TestKeyContext(); TestProcessingTimeService processingTimeService = new TestProcessingTimeService(); HeapInternalTimerService<Integer, String> service = new HeapInternalTimerService<>( testKeyGroupList.getNumberOfKeyGroups(), testKeyGroupList, keyContext, processingTimeService); Assert.assertEquals(startKeyGroupIdx, service.getLocalKeyGroupRangeStartIdx()); }