Repository: apache/kafka
File: clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java (Java)

Code before:
        client.prepareResponse(FindCoordinatorResponse.prepareResponse(error, txnid, node));
    }

    private void prepareInitProducerResponse(Errors error, long producerId, short producerEpoch) {
        client.prepareResponse(initProducerIdResponse(producerId, producerEpoch, error));
    }

    private void assertFutureFailure(Future<?> future, Class<? extends Exception> expectedExceptionType)
            throws InterruptedException {
        assertTrue(future.isDone());
        try {
            future.get();
            fail("Future should have raised " + expectedExceptionType.getName());
        } catch (ExecutionException e) {
            Class<? extends Throwable> causeType = e.getCause().getClass();
            assertTrue(expectedExceptionType.isAssignableFrom(causeType), "Unexpected cause " + causeType.getName());
        }
    }

    private void assertFutureFailure(Future<?> future, Class<? extends Exception> expectedExceptionType,
                                     Class<? extends Exception> rootCauseExceptionType,
                                     String rootCauseExceptionMessage)
            throws InterruptedException {
        assertTrue(future.isDone());
        try {
            future.get();
            fail("Future should have raised " + expectedExceptionType.getName());
        } catch (ExecutionException e) {
            Class<? extends Throwable> causeType = e.getCause().getClass();
            assertTrue(expectedExceptionType.isAssignableFrom(causeType), "Unexpected cause " + causeType.getName());
            assertInstanceOf(rootCauseExceptionType, e.getCause().getCause(), rootCauseExceptionMessage);
        }
    }


    private void createMockClientWithMaxFlightOneMetadataPending() {
        client = new MockClient(time, metadata) {
            volatile boolean canSendMore = true;
            @Override
            public LeastLoadedNode leastLoadedNode(long now) {
                for (Node node : metadata.fetch().nodes()) {
                    if (isReady(node, now) && canSendMore)
                        return new LeastLoadedNode(node, true);
                }
                return new LeastLoadedNode(null, false);
            }

            @Override
            public List<ClientResponse> poll(long timeoutMs, long now) {
                canSendMore = inFlightRequestCount() < 1;
                return super.poll(timeoutMs, now);

Review comment (bug):
would you mind using `assertThrows`?

```java
    private void assertFutureFailure(Future<?> future, Class<? extends Exception> expectedExceptionType,
                                     Class<? extends Exception> rootCauseExceptionType,
                                     String rootCauseExceptionMessage) {
        assertTrue(future.isDone());
        var e = assertThrows(ExecutionException.class, future::get);
        Class<? extends Throwable> causeType = e.getCause().getClass();
        assertTrue(expectedExceptionType.isAssignableFrom(causeType), "Unexpected cause " + causeType.getName());
        assertInstanceOf(rootCauseExceptionType, e.getCause().getCause(), rootCauseExceptionMessage);
    }
```

Code after:
            future.get();
            fail("Future should have raised " + expectedExceptionType.getName());
        } catch (ExecutionException e) {
            Class<? extends Throwable> causeType = e.getCause().getClass();
            assertTrue(expectedExceptionType.isAssignableFrom(causeType), "Unexpected cause " + causeType.getName());
        }
    }

    private void createMockClientWithMaxFlightOneMetadataPending() {
        client = new MockClient(time, metadata) {
            volatile boolean canSendMore = true;
            @Override
            public LeastLoadedNode leastLoadedNode(long now) {
                for (Node node : metadata.fetch().nodes()) {
                    if (isReady(node, now) && canSendMore)
                        return new LeastLoadedNode(node, true);
                }
                return new LeastLoadedNode(null, false);
            }

            @Override
            public List<ClientResponse> poll(long timeoutMs, long now) {
                canSendMore = inFlightRequestCount() < 1;
                return super.poll(timeoutMs, now);
            }
        };

        // Send metadata request and wait until request is sent. `leastLoadedNode` will be null once
        // request is in progress since no more requests can be sent to the node. Node will be ready
        // on the next poll() after response is processed later on in tests which use this method.
        MetadataRequest.Builder builder = new MetadataRequest.Builder(Collections.emptyList(), false);
        Node node = metadata.fetch().nodes().get(0);
        ClientRequest request = client.newClientRequest(node.idString(), builder, time.milliseconds(), true);
        while (!client.ready(node, time.milliseconds()))
            client.poll(0, time.milliseconds());
        client.send(request, time.milliseconds());
        while (client.leastLoadedNode(time.milliseconds()).node() != null)
            client.poll(0, time.milliseconds());
    }

    private void waitForProducerId(TransactionManager transactionManager, ProducerIdAndEpoch producerIdAndEpoch) {
        for (int i = 0; i < 5 && !transactionManager.hasProducerId(); i++)
            sender.runOnce();

        assertTrue(transactionManager.hasProducerId());
        assertEquals(producerIdAndEpoch, transactionManager.producerIdAndEpoch());
    }
    
    private AddPartitionsToTxnResponse buildAddPartitionsToTxnResponseData(int throttleMs, Map<TopicPartition, Errors> errors) {
        AddPartitionsToTxnResponseData.AddPartitionsToTxnResult result = AddPartitionsToTxnResponse.resultForTransaction(
                AddPartitionsToTxnResponse.V3_AND_BELOW_TXN_ID, errors);