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

Code before:

        // cluster authorization errors are fatal, so we should continue seeing it on future sends
        assertTrue(transactionManager.hasFatalError());
        assertSendFailure(ClusterAuthorizationException.class);
    }

    @Test
    public void testCancelInFlightRequestAfterFatalError() throws Exception {
        final long producerId = 343434L;
        TransactionManager transactionManager = createTransactionManager();
        long totalSize = 1024 * 1024;
        String metricGrpName = "producer-custom-metrics";
        MatchingBufferPool pool = new MatchingBufferPool(totalSize, batchSize, metrics, time, metricGrpName);
        setupWithTransactionState(transactionManager, false, pool);

        prepareAndReceiveInitProducerId(producerId, Errors.NONE);
        assertTrue(transactionManager.hasProducerId());

        // cluster authorization is a fatal error for the producer
        Future<RecordMetadata> future1 = appendToAccumulator(tp0);
        sender.runOnce();

        Future<RecordMetadata> future2 = appendToAccumulator(tp1);
        sender.runOnce();

        assertFalse(pool.allMatch());

        client.respond(
            body -> body instanceof ProduceRequest && RequestTestUtils.hasIdempotentRecords((ProduceRequest) body),
            produceResponse(tp0, -1, Errors.CLUSTER_AUTHORIZATION_FAILED, 0));

        sender.runOnce();
        assertTrue(transactionManager.hasFatalError());
        assertFutureFailure(future1, ClusterAuthorizationException.class);

        sender.runOnce();
        assertFutureFailure(future2, ClusterAuthorizationException.class);

        // Should be fine if the second response eventually returns
        client.respond(
            body -> body instanceof ProduceRequest && RequestTestUtils.hasIdempotentRecords((ProduceRequest) body),
            produceResponse(tp1, 0, Errors.NONE, 0));
        sender.runOnce();
        assertTrue(pool.allMatch(), "The batch should have been de-allocated");
    }

    @Test
    public void testUnsupportedForMessageFormatInProduceRequest() throws Exception {
        final long producerId = 343434L;
        TransactionManager transactionManager = createTransactionManager();
        setupWithTransactionState(transactionManager);

Review comment (bug):
Could we do the same check just before the 2nd produce response completes? When I added that, it exposed the bug in RecordAccumulator.

Code after:

        // cluster authorization errors are fatal, so we should continue seeing it on future sends
        assertTrue(transactionManager.hasFatalError());
        assertSendFailure(ClusterAuthorizationException.class);
    }

    @Test
    public void testCancelInFlightRequestAfterFatalError() throws Exception {
        final long producerId = 343434L;
        TransactionManager transactionManager = createTransactionManager();
        long totalSize = 1024 * 1024;
        String metricGrpName = "producer-custom-metrics";
        MatchingBufferPool pool = new MatchingBufferPool(totalSize, batchSize, metrics, time, metricGrpName);
        setupWithTransactionState(transactionManager, false, pool);

        prepareAndReceiveInitProducerId(producerId, Errors.NONE);
        assertTrue(transactionManager.hasProducerId());

        // cluster authorization is a fatal error for the producer
        Future<RecordMetadata> future1 = appendToAccumulator(tp0);
        sender.runOnce();

        Future<RecordMetadata> future2 = appendToAccumulator(tp1);
        sender.runOnce();

        assertFalse(pool.allMatch());

        client.respond(
            body -> body instanceof ProduceRequest && RequestTestUtils.hasIdempotentRecords((ProduceRequest) body),
            produceResponse(tp0, -1, Errors.CLUSTER_AUTHORIZATION_FAILED, 0));

        sender.runOnce();
        assertTrue(transactionManager.hasFatalError());
        assertFutureFailure(future1, ClusterAuthorizationException.class);

        sender.runOnce();
        assertFutureFailure(future2, ClusterAuthorizationException.class);
        assertFalse(pool.allMatch(), "Batch should not be deallocated before the response is received");

        // Should be fine if the second response eventually returns
        client.respond(
            body -> body instanceof ProduceRequest && RequestTestUtils.hasIdempotentRecords((ProduceRequest) body),
            produceResponse(tp1, 0, Errors.NONE, 0));
        sender.runOnce();
        assertTrue(pool.allMatch(), "The batch should have been de-allocated");
    }

    @Test
    public void testUnsupportedForMessageFormatInProduceRequest() throws Exception {
        final long producerId = 343434L;
        TransactionManager transactionManager = createTransactionManager();