WriteKafkaP implements Processor { @Override public void close() { if (snapshotUtility != null) { snapshotUtility.close(); } } private  WriteKafkaP(
            @Nonnull Map<String, Object> properties,
            @Nonnull Function<? super T, ? extends ProducerRecord<K, V>> toRecordFn,
            boolean exactlyOnce
    ); @Override void init(@Nonnull Outbox outbox, @Nonnull Context context); @Override boolean tryProcess(); @SuppressWarnings("unchecked") @Override void process(int ordinal, @Nonnull Inbox inbox); @Override boolean tryProcessWatermark(@Nonnull Watermark watermark); @Override boolean complete(); @Override boolean snapshotCommitPrepare(); @Override boolean snapshotCommitFinish(boolean commitTransactions); @Override void restoreFromSnapshot(@Nonnull Inbox inbox); @Override boolean isCooperative(); @Override void close(); static SupplierEx<Processor> supplier(
            @Nonnull Properties properties,
            @Nonnull Function<? super T, ? extends ProducerRecord<K, V>> toRecordFn,
            boolean exactlyOnce
    ); static final int TXN_POOL_SIZE; }
@Test public void test_resumeTransaction() throws Exception { properties.put("transactional.id", "txn.resumeTransactionTest"); 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); producer.close(); KafkaConsumer<String, String> consumer = kafkaTestSupport.createConsumer(topic); ConsumerRecords<String, String> polledRecords = consumer.poll(Duration.ofSeconds(2)); assertEquals(0, polledRecords.count()); producer = new KafkaProducer<>(properties); ResumeTransactionUtil.resumeTransaction(producer, producerId, epoch, properties.getProperty("transactional.id")); producer.commitTransaction(); StringBuilder actualContents = new StringBuilder(); for (int receivedCount = 0; receivedCount < 2; ) { polledRecords = consumer.poll(Duration.ofSeconds(2)); for (ConsumerRecord<String, 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(); }