@Override
    public void close() {
        if (snapshotUtility != null) {
            snapshotUtility.close();
        }
    }
@Test
    public void test_resumeTransaction() throws Exception {
        properties.setProperty("transactional.id", "txn.resumeTransactionTest");

        // produce items
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
        producer.initTransactions();
        producer.beginTransaction();
        producer.send(new ProducerRecord<>(topic, 0, null, "0")).get();
        producer.send(new ProducerRecord<>(topic, 0, null, "1")).get();
        long producerId = ResumeTransactionUtil.getProducerId(producer);
        short epoch = ResumeTransactionUtil.getEpoch(producer);

        // close the producer immediately to avoid aborting transaction
        producer.close(Duration.ZERO);

        // verify items are not visible
        KafkaConsumer<Integer, String> consumer = kafkaTestSupport.createConsumer(topic);
        ConsumerRecords<Integer, String> polledRecords = consumer.poll(Duration.ofSeconds(2));
        assertEquals(0, polledRecords.count());

        // recover and commit
        producer = new KafkaProducer<>(properties);
        ResumeTransactionUtil.resumeTransaction(producer, producerId, epoch);
        producer.commitTransaction();

        // verify items are visible
        StringBuilder actualContents = new StringBuilder();
        for (int receivedCount = 0; receivedCount < 2; ) {
            polledRecords = consumer.poll(Duration.ofSeconds(2));
            for (ConsumerRecord<Integer, String> record : polledRecords) {
                actualContents.append(record.value()).append('\n');
                receivedCount++;
            }
            logger.info("Received " + receivedCount + " records so far");
        }
        assertEquals("0\n1\n", actualContents.toString());

        producer.close();
        consumer.close();
    }