private boolean isEmpty(ConsumerRecords<K, V> records) {
        return records == null || records.isEmpty();
    }
@Test
    public void when_autoOffsetResetLatest_then_doesNotReadOldMessages() {
        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));

        kafkaTestSupport.produceSync(topic1Name, 0, "0");
        instance().getJet().newJob(p);
        assertTrueAllTheTime(() -> assertTrue(sinkList.isEmpty()), 2);
    }