private boolean isEmpty(ConsumerRecords<K, V> records) {
        return records == null || records.isEmpty();
    }
@Test
    public void when_partitionAddedWhileJobDown_then_consumedFromBeginning() throws Exception {
        String sinkListName = randomName();
        IList<Entry<Integer, String>> sinkList = instance().getList(sinkListName);
        Pipeline p = Pipeline.create();
        Properties properties = properties();
        properties.setProperty("auto.offset.reset", "latest");
        p.readFrom(KafkaSources.<Integer, String>kafka(properties, topic1Name))
                .withoutTimestamps()
                .writeTo(Sinks.list(sinkList));

        Job job = instance().getJet().newJob(p, new JobConfig().setProcessingGuarantee(EXACTLY_ONCE));
        assertTrueEventually(() -> {
            // This might add multiple `0` events to the topic - we need to do this because the source starts from
            // the latest position and we don't exactly know when it starts, so we try repeatedly
            kafkaTestSupport.produce(topic1Name, 0, "0");
            assertFalse(sinkList.isEmpty());
            assertEquals(entry(0, "0"), sinkList.get(0));
        });
        job.suspend();
        assertThat(job).eventuallyHasStatus(SUSPENDED);
        // Note that the job might not have consumed all the zeroes from the topic at this point

        // When
        kafkaTestSupport.setPartitionCount(topic1Name, INITIAL_PARTITION_COUNT + 2);
        // We produce to a partition that didn't exist during the previous job execution.
        // The job must start reading the new partition from the beginning, otherwise it would miss this item.
        Entry<Integer, String> event = produceEventToNewPartition();

        job.resume();
        // All events after the resume will be loaded: the non-consumed zeroes, and the possibly multiple
        // events added in produceEventToNewPartition(). But they must include the event added to the new partition.
        assertTrueEventually(() -> assertThat(sinkList).contains(event));
    }