@Override
  public GenericRow apply(final GenericRow left, final GenericRow right) {

    final GenericRow row = new GenericRow(leftCount + rightCount + additionalCount);

    if (left != null) {
      row.appendAll(left.values());
    } else {
      fillWithNulls(row, leftCount);
    }

    if (right != null) {
      row.appendAll(right.values());
    } else {
      fillWithNulls(row, rightCount);
    }

    // Potentially append additional nulls as a holder for a synthetic key columns.
    // These columns are not populated, as they are not accessed, but must be present for the row
    // to match the rows schema.
    fillWithNulls(row, additionalCount);

    return row;
  }
@Test
  public void shouldJoinValueBothNonNull() {
    final KsqlValueJoiner joiner = new KsqlValueJoiner(leftSchema.value().size(),
        rightSchema.value().size(), 0
    );

    final GenericRow joined = joiner.apply(leftRow, rightRow);
    final List<Object> expected = Arrays.asList(12L, "foobar", 20L, "baz");
    assertEquals(expected, joined.values());
  }