flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/HeapInternalTimerService.java
/**
	 * Restore the timers (both processing and event time ones) for a given {@code keyGroupIdx}.
	 *
	 * @param restoredTimersSnapshot the restored snapshot containing the key-group's timers,
	 *                       and the serializers that were used to write them
	 * @param keyGroupIdx the id of the key-group to be put in the snapshot.
	 */
Restore the timers (both processing and event time ones) for a given  keyGroupIdx.
@param restoredTimersSnapshot the restored snapshot containing the key-group's timers,
and the serializers that were used to write them
@param keyGroupIdx the id of the key-group to be put in the snapshot.
Restore the timers ( both processing and event time ones ) for a given keyGroupIdx . @param restoredTimersSnapshot the restored snapshot containing the key - group ' s timers and the serializers that were used to write them @param keyGroupIdx the id of the key - group to be put in the snapshot .
@SuppressWarnings("unchecked")
	public void restoreTimersForKeyGroup(InternalTimersSnapshot<?, ?> restoredTimersSnapshot, int keyGroupIdx) throws IOException {
		this.restoredTimersSnapshot = (InternalTimersSnapshot<K, N>) restoredTimersSnapshot;
		if ((this.keyDeserializer != null && !this.keyDeserializer.equals(restoredTimersSnapshot.getKeySerializer())) ||
			(this.namespaceDeserializer != null && !this.namespaceDeserializer.equals(restoredTimersSnapshot.getNamespaceSerializer()))) {
			throw new IllegalArgumentException("Tried to restore timers " +
				"for the same service with different serializers.");
		}
		this.keyDeserializer = this.restoredTimersSnapshot.getKeySerializer();
		this.namespaceDeserializer = this.restoredTimersSnapshot.getNamespaceSerializer();
		checkArgument(localKeyGroupRange.contains(keyGroupIdx),
			"Key Group " + keyGroupIdx + " does not belong to the local range.");
		Set<InternalTimer<K, N>> eventTimers = getEventTimeTimerSetForKeyGroup(keyGroupIdx);
		eventTimers.addAll(this.restoredTimersSnapshot.getEventTimeTimers());
		eventTimeTimersQueue.addAll(this.restoredTimersSnapshot.getEventTimeTimers());
		Set<InternalTimer<K, N>> processingTimers = getProcessingTimeTimerSetForKeyGroup(keyGroupIdx);
		processingTimers.addAll(this.restoredTimersSnapshot.getProcessingTimeTimers());
		processingTimeTimersQueue.addAll(this.restoredTimersSnapshot.getProcessingTimeTimers());
	}
@ SuppressWarnings ( "unchecked" ) public void restoreTimersForKeyGroup ( InternalTimersSnapshot < ? , ? > restoredTimersSnapshot , int keyGroupIdx ) throws IOException { this . restoredTimersSnapshot = ( InternalTimersSnapshot < K , N > ) restoredTimersSnapshot ; if ( ( this . keyDeserializer != null && ! this . keyDeserializer . equals ( restoredTimersSnapshot . getKeySerializer ( ) ) ) || ( this . namespaceDeserializer != null && ! this . namespaceDeserializer . equals ( restoredTimersSnapshot . getNamespaceSerializer ( ) ) ) ) { throw new IllegalArgumentException ( "Tried to restore timers " + "for the same service with different serializers." ) ; } this . keyDeserializer = this . restoredTimersSnapshot . getKeySerializer ( ) ; this . namespaceDeserializer = this . restoredTimersSnapshot . getNamespaceSerializer ( ) ; checkArgument ( localKeyGroupRange . contains ( keyGroupIdx ) , "Key Group " + keyGroupIdx + " does not belong to the local range." ) ; Set < InternalTimer < K , N > > eventTimers = getEventTimeTimerSetForKeyGroup ( keyGroupIdx ) ; eventTimers . addAll ( this . restoredTimersSnapshot . getEventTimeTimers ( ) ) ; eventTimeTimersQueue . addAll ( this . restoredTimersSnapshot . getEventTimeTimers ( ) ) ; Set < InternalTimer < K , N > > processingTimers = getProcessingTimeTimerSetForKeyGroup ( keyGroupIdx ) ; processingTimers . addAll ( this . restoredTimersSnapshot . getProcessingTimeTimers ( ) ) ; processingTimeTimersQueue . addAll ( this . restoredTimersSnapshot . getProcessingTimeTimers ( ) ) ; }
Restore the timers (both processing and event time ones) for a given {@code keyGroupIdx}.
Restore the timers ( both processing and event time ones ) for a given { @code keyGroupIdx } .
// restore the event time timers // restore the processing time timers