@Override
    protected void init(@Nonnull Context context) {
        topics = new ArrayList<>(topicsConfig.getTopicNames());
        for (String topic : topics) {
            offsets.put(topic, new long[0]);
        }
        processorIndex = context.globalProcessorIndex();
        totalParallelism = context.totalParallelism();
        processingGuarantee = context.processingGuarantee();
        consumer = kafkaConsumerFn.apply(context);
        if (processingGuarantee == NONE) {
            warnWhenInitialOffsetsProvided();
        }
    }
@Test
    public void when_customProjection_then_used() throws Exception {
        // When
        var processor = createProcessor(properties(), 2, r -> r.key() + "=" + r.value(), 10_000);
        TestOutbox outbox = new TestOutbox(new int[]{10}, 10);
        processor.init(outbox, new TestProcessorContext());
        kafkaTestSupport.produceSync(topic1Name, 0, "0");

        // Then
        assertEquals("0=0", consumeEventually(processor, outbox));
    }