code: public void snapshotStateForKeyGroup(DataOutputView stream, int keyGroupIdx) throws IOException {
  		Preconditions.checkState(useLegacySynchronousSnapshots);
  		InternalTimerServiceSerializationProxy<K> serializationProxy =
  			new InternalTimerServiceSerializationProxy<>(this, keyGroupIdx);
  
  		serializationProxy.write(stream);
  	}
docs: ////////////////				Fault Tolerance Methods				///////////////////