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
/**
	 * Starts the local {@link HeapInternalTimerService} by:
	 * <ol>
	 *     <li>Setting the {@code keySerialized} and {@code namespaceSerializer} for the timers it will contain.</li>
	 *     <li>Setting the {@code triggerTarget} which contains the action to be performed when a timer fires.</li>
	 *     <li>Re-registering timers that were retrieved after recoveting from a node failure, if any.</li>
	 * </ol>
	 * This method can be called multiple times, as long as it is called with the same serializers.
	 */
Starts the local  HeapInternalTimerService by:

Setting the  keySerialized and  namespaceSerializer for the timers it will contain.
Setting the  triggerTarget which contains the action to be performed when a timer fires.
Re-registering timers that were retrieved after recoveting from a node failure, if any.

This method can be called multiple times, as long as it is called with the same serializers.
Starts the local HeapInternalTimerService by : Setting the keySerialized and namespaceSerializer for the timers it will contain . Setting the triggerTarget which contains the action to be performed when a timer fires . Re - registering timers that were retrieved after recoveting from a node failure if any . This method can be called multiple times as long as it is called with the same serializers .
public void startTimerService(
			TypeSerializer<K> keySerializer,
			TypeSerializer<N> namespaceSerializer,
			Triggerable<K, N> triggerTarget) {
		if (!isInitialized) {
			if (keySerializer == null || namespaceSerializer == null) {
				throw new IllegalArgumentException("The TimersService serializers cannot be null.");
			}
			if (this.keySerializer != null || this.namespaceSerializer != null || this.triggerTarget != null) {
				throw new IllegalStateException("The TimerService has already been initialized.");
			}
			if (restoredTimersSnapshot != null) {
				CompatibilityResult<K> keySerializerCompatibility = CompatibilityUtil.resolveCompatibilityResult(
					this.keyDeserializer,
					null,
					restoredTimersSnapshot.getKeySerializerConfigSnapshot(),
					keySerializer);
				CompatibilityResult<N> namespaceSerializerCompatibility = CompatibilityUtil.resolveCompatibilityResult(
					this.namespaceDeserializer,
					null,
					restoredTimersSnapshot.getNamespaceSerializerConfigSnapshot(),
					namespaceSerializer);
				if (keySerializerCompatibility.isRequiresMigration() || namespaceSerializerCompatibility.isRequiresMigration()) {
					throw new IllegalStateException("Tried to initialize restored TimerService " +
						"with incompatible serializers than those used to snapshot its state.");
				}
			}
			this.keySerializer = keySerializer;
			this.namespaceSerializer = namespaceSerializer;
			this.keyDeserializer = null;
			this.namespaceDeserializer = null;
			this.triggerTarget = Preconditions.checkNotNull(triggerTarget);
			if (processingTimeTimersQueue.size() > 0) {
				nextTimer = processingTimeService.registerTimer(processingTimeTimersQueue.peek().getTimestamp(), this);
			}
			this.isInitialized = true;
		} else {
			if (!(this.keySerializer.equals(keySerializer) && this.namespaceSerializer.equals(namespaceSerializer))) {
				throw new IllegalArgumentException("Already initialized Timer Service " +
					"tried to be initialized with different key and namespace serializers.");
			}
		}
	}
public void startTimerService ( TypeSerializer < K > keySerializer , TypeSerializer < N > namespaceSerializer , Triggerable < K , N > triggerTarget ) { if ( ! isInitialized ) { if ( keySerializer == null || namespaceSerializer == null ) { throw new IllegalArgumentException ( "The TimersService serializers cannot be null." ) ; } if ( this . keySerializer != null || this . namespaceSerializer != null || this . triggerTarget != null ) { throw new IllegalStateException ( "The TimerService has already been initialized." ) ; } if ( restoredTimersSnapshot != null ) { CompatibilityResult < K > keySerializerCompatibility = CompatibilityUtil . resolveCompatibilityResult ( this . keyDeserializer , null , restoredTimersSnapshot . getKeySerializerConfigSnapshot ( ) , keySerializer ) ; CompatibilityResult < N > namespaceSerializerCompatibility = CompatibilityUtil . resolveCompatibilityResult ( this . namespaceDeserializer , null , restoredTimersSnapshot . getNamespaceSerializerConfigSnapshot ( ) , namespaceSerializer ) ; if ( keySerializerCompatibility . isRequiresMigration ( ) || namespaceSerializerCompatibility . isRequiresMigration ( ) ) { throw new IllegalStateException ( "Tried to initialize restored TimerService " + "with incompatible serializers than those used to snapshot its state." ) ; } } this . keySerializer = keySerializer ; this . namespaceSerializer = namespaceSerializer ; this . keyDeserializer = null ; this . namespaceDeserializer = null ; this . triggerTarget = Preconditions . checkNotNull ( triggerTarget ) ; if ( processingTimeTimersQueue . size ( ) > 0 ) { nextTimer = processingTimeService . registerTimer ( processingTimeTimersQueue . peek ( ) . getTimestamp ( ) , this ) ; } this . isInitialized = true ; } else { if ( ! ( this . keySerializer . equals ( keySerializer ) && this . namespaceSerializer . equals ( namespaceSerializer ) ) ) { throw new IllegalArgumentException ( "Already initialized Timer Service " + "tried to be initialized with different key and namespace serializers." ) ; } } }
Starts the local {@link HeapInternalTimerService} by:
 <ol>
     <li>Setting the {@code keySerialized} and {@code namespaceSerializer} for the timers it will contain.</li>
     <li>Setting the {@code triggerTarget} which contains the action to be performed when a timer fires.</li>
     <li>Re-registering timers that were retrieved after recoveting from a node failure, if any.</li>
 </ol>
 This method can be called multiple times, as long as it is called with the same serializers.
Starts the local { @link HeapInternalTimerService } by : <ol > <li > Setting the { @code keySerialized } and { @code namespaceSerializer } for the timers it will contain . < / li > <li > Setting the { @code triggerTarget } which contains the action to be performed when a timer fires . < / li > <li > Re - registering timers that were retrieved after recoveting from a node failure if any . < / li > < / ol > This method can be called multiple times as long as it is called with the same serializers .
// the following is the case where we restore // re-register the restored timers (if any)