private Collection<JM> getJoinOutput(K key, Object value, M message) {
    R record = value == null ? null : (R) KV.of(key, value);

    JM output = joinOpSpec.getJoinFn().apply(message, record);

    // The support for inner and outer join will be provided in the jonFn. For inner join, the joinFn might
    // return null, when the corresponding record is absent in the table.
    return output != null ?
        Collections.singletonList(output) : Collections.emptyList();
  }
Implementation of a stream-table join operator that first retrieve the value of
the message key from incoming message, and then apply the join function.

@param <K> type of the join key
@param <M> type of input messages
@param <R> type of the table record
@param <JM> type of the join result
samza-core/src/main/java/org/apache/samza/operators/impl/StreamTableJoinOperatorImpl.java