@Test
    public void testSequenceNumberIncrement() throws InterruptedException {
        final long producerId = 343434L;
        TransactionManager transactionManager = createTransactionManager();
        setupWithTransactionState(transactionManager);
        prepareAndReceiveInitProducerId(producerId, Errors.NONE);
        assertTrue(transactionManager.hasProducerId());

        int maxRetries = 10;
        Metrics m = new Metrics();
        SenderMetricsRegistry senderMetrics = new SenderMetricsRegistry(m);

        Sender sender = new Sender(logContext, client, metadata, this.accumulator, true, MAX_REQUEST_SIZE, ACKS_ALL, maxRetries,
                senderMetrics, time, REQUEST_TIMEOUT, RETRY_BACKOFF_MS, transactionManager, apiVersions);

        Future<RecordMetadata> responseFuture = appendToAccumulator(tp0);
        client.prepareResponse(body -> {
            if (body instanceof ProduceRequest) {
                ProduceRequest request = (ProduceRequest) body;
                MemoryRecords records = partitionRecords(request).get(tp0);
                Iterator<MutableRecordBatch> batchIterator = records.batches().iterator();
                assertTrue(batchIterator.hasNext());
                RecordBatch batch = batchIterator.next();
                assertFalse(batchIterator.hasNext());
                assertEquals(0, batch.baseSequence());
                assertEquals(producerId, batch.producerId());
                assertEquals(0, batch.producerEpoch());
                return true;
            }
            return false;
        }, produceResponse(tp0, 0, Errors.NONE, 0));

        sender.runOnce();  // connect.
        sender.runOnce();  // send.

        sender.runOnce();  // receive response
        assertTrue(responseFuture.isDone());
        assertEquals(OptionalInt.of(0), transactionManager.lastAckedSequence(tp0));
        assertEquals(1L, transactionManager.sequenceNumber(tp0));
    }
Tests the code path where the target node to send FindCoordinator or InitProducerId
is not ready.
clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java