@Override
      public void schedule(K key, long time) {
        callbackScheduler.scheduleCallback(key, time, (k, collector, coordinator) -> {
          final ScheduledFunction<K, RM> scheduledFn = getOperatorSpec().getScheduledFn();
          if (scheduledFn != null) {
            final Collection<RM> output = scheduledFn.onCallback(key, time);

            if (!output.isEmpty()) {
              CompletableFuture<Void> timerFuture = CompletableFuture.allOf(output.stream()
                  .flatMap(r -> registeredOperators.stream()
                      .map(op -> op.onMessageAsync(r, collector, coordinator)))
                  .toArray(CompletableFuture[]::new));

              timerFuture.join();
            }
          } else {
            throw new SamzaException(
                String.format("Operator %s id %s (created at %s) must implement ScheduledFunction to use system timer.",
                    getOperatorSpec().getOpCode().name(), getOpImplId(), getOperatorSpec().getSourceLocation()));
          }
        });
      }
Returns a registry which allows registering arbitrary system-clock timer with K-typed key.
The user-defined function in the operator spec needs to implement {@link ScheduledFunction#onCallback(Object, long)}
for timer notifications.
@param <K> key type for the timer.
@return an instance of {@link Scheduler}
samza-core/src/main/java/org/apache/samza/operators/impl/OperatorImpl.java