@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_eventsInSinglePartition_then_watermarkAfterIdleTime() throws Exception {
        // When
        var processor = createProcessor(properties(), 2, r -> entry(r.key(), r.value()), 10_000);
        TestOutbox outbox = new TestOutbox(new int[]{10}, 10);
        processor.init(outbox, new TestProcessorContext());
        kafkaTestSupport.produceSync(topic1Name, 10, "foo");

        // Then
        assertEquals(entry(10, "foo"), consumeEventually(processor, outbox));
        long time1 = System.nanoTime();
        assertEquals(new Watermark(10 - LAG), consumeEventually(processor, outbox));
        long time2 = System.nanoTime();
        long elapsedMs = NANOSECONDS.toMillis(time2 - time1);
        assertBetween("elapsed time", elapsedMs, 3000, 30_000);
    }