@Override
public InternalTimerService<VoidNamespace> createTimerService(
    Triggerable<String, VoidNamespace> triggerable) {
  final TimerSerializer<String, VoidNamespace> timerSerializer =
      new TimerSerializer<>(StringSerializer.INSTANCE, VoidNamespaceSerializer.INSTANCE);

  return timeServiceManager.getInternalTimerService(
      DELAYED_MSG_TIMER_SERVICE_NAME, timerSerializer, triggerable);
}
 

@Override
public InternalTimerService<VoidNamespace> createTimerService(
    Triggerable<String, VoidNamespace> triggerable) {
  final TimerSerializer<String, VoidNamespace> timerSerializer =
      new TimerSerializer<>(StringSerializer.INSTANCE, VoidNamespaceSerializer.INSTANCE);

  return timeServiceManager.getInternalTimerService(
      DELAYED_MSG_TIMER_SERVICE_NAME, timerSerializer, triggerable);
}
 

@SuppressWarnings("unchecked")
private InternalTimerService<VoidNamespace> restoreTimerService(StreamOperatorStateContext context) {
	InternalTimeServiceManager<K> timeServiceManager = (InternalTimeServiceManager<K>) context.internalTimerServiceManager();
	TimerSerializer<K, VoidNamespace> timerSerializer = new TimerSerializer<>(keySerializer, VoidNamespaceSerializer.INSTANCE);
	return timeServiceManager.getInternalTimerService(USER_TIMERS_NAME, timerSerializer, VoidTriggerable.instance());
}
 

protected final InternalTimerService<N> getInternalTimerService(String name) {
	TimerSerializer<KEY, N> timerSerializer = new TimerSerializer<>(keySerializer, namespaceSerializer);
	return timerServiceManager.getInternalTimerService(name, timerSerializer, VoidTriggerable.instance());
}