@SuppressWarnings("unchecked")
	public void restoreTimersForKeyGroup(InternalTimersSnapshot<?, ?> restoredSnapshot, int keyGroupIdx) {
		this.restoredTimersSnapshot = (InternalTimersSnapshot<K, N>) restoredSnapshot;

		TypeSerializer<K> restoredKeySerializer = restoredTimersSnapshot.getKeySerializerSnapshot().restoreSerializer();
		if (this.keyDeserializer != null && !this.keyDeserializer.equals(restoredKeySerializer)) {
			throw new IllegalArgumentException("Tried to restore timers for the same service with different key serializers.");
		}
		this.keyDeserializer = restoredKeySerializer;

		TypeSerializer<N> restoredNamespaceSerializer = restoredTimersSnapshot.getNamespaceSerializerSnapshot().restoreSerializer();
		if (this.namespaceDeserializer != null && !this.namespaceDeserializer.equals(restoredNamespaceSerializer)) {
			throw new IllegalArgumentException("Tried to restore timers for the same service with different namespace serializers.");
		}
		this.namespaceDeserializer = restoredNamespaceSerializer;

		checkArgument(localKeyGroupRange.contains(keyGroupIdx),
			"Key Group " + keyGroupIdx + " does not belong to the local range.");

		// restore the event time timers
		eventTimeTimersQueue.addAll(restoredTimersSnapshot.getEventTimeTimers());

		// restore the processing time timers
		processingTimeTimersQueue.addAll(restoredTimersSnapshot.getProcessingTimeTimers());
	}
@ SuppressWarnings ( "unchecked" ) public void restoreTimersForKeyGroup ( InternalTimersSnapshot < ? , ? > restoredSnapshot , int keyGroupIdx ) { this . restoredTimersSnapshot = ( InternalTimersSnapshot < K , N > ) restoredSnapshot ; TypeSerializer < K > restoredKeySerializer = restoredTimersSnapshot . getKeySerializerSnapshot ( ) . restoreSerializer ( ) ; if ( this . keyDeserializer != null && ! this . keyDeserializer . equals ( restoredKeySerializer ) ) { throw new IllegalArgumentException ( "Tried to restore timers for the same service with different key serializers." ) ; } this . keyDeserializer = restoredKeySerializer ; TypeSerializer < N > restoredNamespaceSerializer = restoredTimersSnapshot . getNamespaceSerializerSnapshot ( ) . restoreSerializer ( ) ; if ( this . namespaceDeserializer != null && ! this . namespaceDeserializer . equals ( restoredNamespaceSerializer ) ) { throw new IllegalArgumentException ( "Tried to restore timers for the same service with different namespace serializers." ) ; } this . namespaceDeserializer = restoredNamespaceSerializer ; checkArgument ( localKeyGroupRange . contains ( keyGroupIdx ) , "Key Group " + keyGroupIdx + " does not belong to the local range." ) ; eventTimeTimersQueue . addAll ( restoredTimersSnapshot . getEventTimeTimers ( ) ) ; processingTimeTimersQueue . addAll ( restoredTimersSnapshot . getProcessingTimeTimers ( ) ) ; }
Restore the timers (both processing and event time ones) for a given {@code keyGroupIdx}.

@param restoredSnapshot 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.