@Override
  public SamzaSqlRelMessage apply(SamzaSqlRelMessage message, R record) {

    if (joinRelType.compareTo(JoinRelType.INNER) == 0 && record == null) {
      log.debug("Inner Join: Record not found for the message with key: " + getMessageKey(message));
      // Returning null would result in Join operator implementation to filter out the message.
      return null;
    }

    // The resulting join output should be a SamzaSqlRelMessage containing the fields from both the stream message and
    // table record. The order of stream message fields and table record fields are dictated by the position of stream
    // and table in the 'from' clause of sql query. The output should also include the keys from both the stream message
    // and the table record.
    List<Object> outFieldValues = new ArrayList<>();

    // If table position is on the right, add the stream message fields first
    if (isTablePosOnRight) {
      outFieldValues.addAll(message.getSamzaSqlRelRecord().getFieldValues());
    }

    // Add the table record fields.
    if (record != null) {
      outFieldValues.addAll(getTableRelRecordFieldValues(record));
    } else {
      // Table record could be null as the record could not be found in the store. This can
      // happen for outer joins. Add nulls to all the field values in the output message.
      tableFieldNames.forEach(s -> outFieldValues.add(null));
    }

    // If table position is on the left, add the stream message fields last
    if (!isTablePosOnRight) {
      outFieldValues.addAll(message.getSamzaSqlRelRecord().getFieldValues());
    }

    return new SamzaSqlRelMessage(outFieldNames, outFieldValues, message.getSamzaSqlRelMsgMetadata());
  }
This abstract class joins incoming {@link SamzaSqlRelMessage} with records from a table with the join key.
samza-sql/src/main/java/org/apache/samza/sql/translator/SamzaSqlTableJoinFunction.java