@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_eventsInAllPartitions_then_watermarkOutputImmediately() throws Exception {
        var processor = createProcessor(properties(), 1, r -> entry(r.key(), r.value()), 10_000);
        TestOutbox outbox = new TestOutbox(new int[]{10}, 10);
        processor.init(outbox, new TestProcessorContext());

        for (int i = 0; i < INITIAL_PARTITION_COUNT; i++) {
            Entry<Integer, String> event = entry(i + 100, Integer.toString(i));
            System.out.println("produced event " + event);
            //Wait for the event to be published to Kafka, the processor can access Kafka metadata
            kafkaTestSupport.produce(topic1Name, i, null, event.getKey(), event.getValue()).get();
            if (i == INITIAL_PARTITION_COUNT - 1) {
                assertEquals(new Watermark(100 - LAG), consumeEventually(processor, outbox));
            }
            assertEquals(event, consumeEventually(processor, outbox));
        }
    }