@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_partitionAddedWhilePartitionsInitialOffsetsProvided_then_consumedFromBeginning() throws Exception {
        Properties properties = properties();
        properties.setProperty("metadata.max.age.ms", "100");
        TopicsConfig topicsConfig = new TopicsConfig()
                .addTopic(topic2Name)
                .addTopicConfig(new TopicConfig(topic1Name)
                        .addPartitionInitialOffset(0, 1L)
                        .addPartitionInitialOffset(1, 1L)
                        .addPartitionInitialOffset(2, 1L)
                        .addPartitionInitialOffset(3, 1L)
                        // specify initial offset for non-existing partitions as well
                        .addPartitionInitialOffset(4, 1L)
                        .addPartitionInitialOffset(5, 1L)
                );

        StreamKafkaP<Integer, String, Entry<Integer, String>> processor = createProcessor(
                properties, topicsConfig, r -> entry(r.key(), r.value()), 60_000);
        TestOutbox outbox = new TestOutbox(new int[]{10}, 10);
        TestProcessorContext context = new TestProcessorContext();
        context.setProcessingGuarantee(AT_LEAST_ONCE);
        processor.init(outbox, context);

        kafkaTestSupport.produceSync(topic1Name, 0, "0"); // first record will be skipped due to topics config
        kafkaTestSupport.produceSync(topic1Name, 1, "1");
        assertEquals(entry(1, "1"), consumeEventually(processor, outbox));

        kafkaTestSupport.setPartitionCount(topic1Name, INITIAL_PARTITION_COUNT + 2);

        boolean somethingInPartition1 = false;
        for (int i = 2; i < 12; i++) {
            Future<RecordMetadata> future = kafkaTestSupport.produce(topic1Name, i, Integer.toString(i));
            RecordMetadata recordMetadata = future.get();
            System.out.println("## Entry " + i + " produced to partition " + recordMetadata.partition());
            somethingInPartition1 |= recordMetadata.partition() == 1;
        }
        assertTrue("nothing was produced to partition-1", somethingInPartition1);
        Set<Object> receivedEvents = new LinkedHashSet<>();
        for (int i = 2; i < 12;) {
            try {
                Object consumed = consumeEventually(processor, outbox);
                if (!(consumed instanceof Watermark)) {
                    receivedEvents.add(consumed);
                    i++;
                }
            } catch (AssertionError e) {
                throw new AssertionError("Unable to receive 10 items, events so far: " + receivedEvents, e);
            }
        }
        var expected = range(2, 12).mapToObj(i -> entry(i, Integer.toString(i))).collect(toSet());
        assertThat(receivedEvents).containsExactlyInAnyOrderElementsOf(expected);
    }