@Override
    public Processor<K, Change<V1>, K, Change<VOut>> get() {
        return new KTableKTableOuterJoinProcessor(valueGetterSupplier2.get());
    }
@Test
    public void shouldLogAndMeterSkippedRecordsDueToNullLeftKey() {
        final StreamsBuilder builder = new StreamsBuilder();

        @SuppressWarnings("unchecked")
        final Processor<String, Change<String>, String, Change<Object>> join = new KTableKTableOuterJoin<>(
                (KTableImpl<String, String, String>) builder.table("left", Consumed.with(Serdes.String(), Serdes.String())),
                (KTableImpl<String, String, String>) builder.table("right", Consumed.with(Serdes.String(), Serdes.String())),
                null
        ).get();

        final MockProcessorContext<String, Change<Object>> context = new MockProcessorContext<>(props);
        context.setRecordMetadata("left", -1, -2);
        join.init(context);

        try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(KTableKTableOuterJoin.class)) {
            join.process(new Record<>(null, new Change<>("new", "old"), 0));

            assertThat(
                appender.getMessages(),
                hasItem("Skipping record due to null key. topic=[left] partition=[-1] offset=[-2]")
            );
        }
    }