@Override
  protected CompletionStage<Collection<JM>> handleMessageAsync(M message, MessageCollector collector,
      TaskCoordinator coordinator) {
    Collection<JM> output = Collections.emptyList();

    try {
      KeyValueStore<K, TimestampedValue<M>> thisState = thisPartialJoinFn.getState();
      KeyValueStore<K, TimestampedValue<OM>> otherState = otherPartialJoinFn.getState();

      K key = thisPartialJoinFn.getKey(message);
      thisState.put(key, new TimestampedValue<>(message, clock.currentTimeMillis()));
      TimestampedValue<OM> otherMessage = otherState.get(key);

      long now = clock.currentTimeMillis();
      if (otherMessage != null && otherMessage.getTimestamp() > now - ttlMs) {
        JM joinResult = thisPartialJoinFn.apply(message, otherMessage.getValue());
        output = Collections.singletonList(joinResult);
      }
    } catch (Exception e) {
      throw new SamzaException("Error handling message in PartialJoinOperatorImpl " + getOpImplId(), e);
    }

    return CompletableFuture.completedFuture(output);
  }
Implementation of one side of a {@link JoinOperatorSpec} that buffers and joins its input messages of
type {@code M} with buffered input messages of type {@code OM} in the paired {@link PartialJoinOperatorImpl}.

@param <K> the type of join key
@param <M> the type of input messages on this side of the join
@param <OM> the type of input message on the other side of the join
@param <JM> the type of join result
samza-core/src/main/java/org/apache/samza/operators/impl/PartialJoinOperatorImpl.java