@Test
    public void testWhenProduceResponseReturnsWithALeaderShipChangeErrorAndNewLeaderInformation()
        throws InterruptedException {
        // Setup 3 partitions, tp0 & tp1 return with NOT_LEADER_OR_FOLLOWER, tp2 doesn't return an error.
        Metrics m = new Metrics();
        SenderMetricsRegistry senderMetrics = new SenderMetricsRegistry(m);
        try {
            // SETUP
            String metricGrpName = "producer-metrics-test-stats-1";
            long totalSize = 1024 * 1024;
            BufferPool pool = new BufferPool(totalSize, batchSize, metrics, time,
                metricGrpName);
            long retryBackoffMaxMs = 100L;
            // lingerMs is 0 to send batch as soon as any records are available on it.
            this.accumulator = new RecordAccumulator(logContext, batchSize,
                Compression.NONE, 0, 10L, retryBackoffMaxMs,
                DELIVERY_TIMEOUT_MS, metrics, metricGrpName, time, apiVersions, null, pool);
            Sender sender = new Sender(logContext, client, metadata, this.accumulator, false,
                MAX_REQUEST_SIZE, ACKS_ALL,
                10, senderMetrics, time, REQUEST_TIMEOUT, RETRY_BACKOFF_MS, null,
                apiVersions);
            // Update metadata with leader-epochs.
            int tp0LeaderEpoch = 100;
            int tp1LeaderEpoch = 200;
            int tp2LeaderEpoch = 300;
            this.client.updateMetadata(
                RequestTestUtils.metadataUpdateWith(1, Collections.singletonMap("test", 3),
                    tp -> {
                        if (tp0.equals(tp)) {
                            return tp0LeaderEpoch;
                        }  else if (tp1.equals(tp)) {
                            return tp1LeaderEpoch;
                        } else if (tp2.equals(tp)) {
                            return tp2LeaderEpoch;
                        } else {
                            throw new RuntimeException("unexpected tp " + tp);
                        }
                    }));
            Cluster startingMetadataCluster = metadata.fetch();

            // Produce to tp0/1/2, where NO_LEADER_OR_FOLLOWER with new leader info is returned for tp0/1, and tp2 is returned without errors.
            Future<RecordMetadata> futureIsProducedTp0 = appendToAccumulator(tp0, 0L, "key", "value");
            Future<RecordMetadata> futureIsProducedTp1 = appendToAccumulator(tp1, 0L, "key", "value");
            Future<RecordMetadata> futureIsProducedTp2 = appendToAccumulator(tp2, 0L, "key", "value");
            sender.runOnce(); // connect
            sender.runOnce(); // send produce request
            assertEquals(1, sender.inFlightBatches(tp0).size());
            assertEquals(1, sender.inFlightBatches(tp1).size());
            assertEquals(1, sender.inFlightBatches(tp2).size());
            assertTrue(client.hasInFlightRequests());

            Node newNodeForTp0 = new Node(9990, "newhost9990", 9990, "newrack9990");
            Node newNodeForTp1 = new Node(9991, "newhost9991", 9991, "newrack9991");
            List<Node> newNodes = Arrays.asList(newNodeForTp0, newNodeForTp1);

            Map<TopicPartition, OffsetAndError> responses = new LinkedHashMap<>();
            responses.put(tp0, new OffsetAndError(-1, Errors.NOT_LEADER_OR_FOLLOWER));
            responses.put(tp1, new OffsetAndError(-1, Errors.NOT_LEADER_OR_FOLLOWER));
            responses.put(tp2, new OffsetAndError(100, Errors.NONE));

            Map<TopicPartition, ProduceResponseData.LeaderIdAndEpoch> partitionLeaderInfo = new HashMap<>();
            ProduceResponseData.LeaderIdAndEpoch tp0LeaderInfo = new ProduceResponseData.LeaderIdAndEpoch();
            tp0LeaderInfo.setLeaderEpoch(tp0LeaderEpoch + 1);
            tp0LeaderInfo.setLeaderId(newNodeForTp0.id());
            partitionLeaderInfo.put(tp0, tp0LeaderInfo);
            ProduceResponseData.LeaderIdAndEpoch tp1LeaderInfo = new ProduceResponseData.LeaderIdAndEpoch();
            tp1LeaderInfo.setLeaderEpoch(tp1LeaderEpoch + 1);
            tp1LeaderInfo.setLeaderId(newNodeForTp1.id());
            partitionLeaderInfo.put(tp1, tp1LeaderInfo);

            client.respond(produceResponse(responses, partitionLeaderInfo, newNodes));
            sender.runOnce(); // receive produce response, batch scheduled for retry
            assertFalse(futureIsProducedTp0.isDone(), "Produce request to tp0 should be unfinished.");
            assertFalse(futureIsProducedTp1.isDone(), "Produce request to tp1 should be unfinished.");
            assertTrue(futureIsProducedTp2.isDone(), "Produce request to tp0 should be done.");

            // Validate metadata is unchanged as new leader info wasn't received.
            assertNotEquals(startingMetadataCluster, metadata.fetch());
            // Validate metadata cached has updated leader info for tp0/1.
            Metadata.LeaderAndEpoch tp0NewLeaderInfo = metadata.currentLeader(tp0);
            assertEquals(newNodeForTp0, tp0NewLeaderInfo.leader.get());
            assertEquals(tp0LeaderEpoch + 1, tp0NewLeaderInfo.epoch.get());
            Metadata.LeaderAndEpoch tp1NewLeaderInfo = metadata.currentLeader(tp1);
            assertEquals(newNodeForTp1, tp1NewLeaderInfo.leader.get());
            assertEquals(tp1LeaderEpoch + 1, tp1NewLeaderInfo.epoch.get());

            // Validate metadata-refresh is requested as NOT_LEADER_OR_FOLLOWER received earlier
            assertTrue(metadata.updateRequested());

            // TEST that a subsequent retry skips the backoff as a new leader information is available.
            sender.runOnce(); //send produce request
            assertEquals(1, sender.inFlightBatches(tp0).size());
            assertEquals(1, sender.inFlightBatches(tp1).size());
            assertTrue(client.hasInFlightRequests());
        } finally {
            m.close();
        }
    }
Test the scenario that FetchResponse returns NOT_LEADER_OR_FOLLOWER, indicating change in leadership, along with
new leader info(defined in KIP-951).
clients/src/test/java/org/apache/kafka/clients/producer/internals/SenderTest.java