private MockClient.RequestMatcher produceRequestMatcher(final TopicPartition tp,
                                                            final ProducerIdAndEpoch producerIdAndEpoch,
                                                            final int sequence,
                                                            final boolean isTransactional) {
        return body -> {
            if (!(body instanceof ProduceRequest))
                return false;

            ProduceRequest request = (ProduceRequest) body;
            Map<TopicPartition, MemoryRecords> recordsMap = partitionRecords(request);
            MemoryRecords records = recordsMap.get(tp);
            if (records == null)
                return false;

            List<MutableRecordBatch> batches = TestUtils.toList(records.batches());
            if (batches.size() != 1)
                return false;

            MutableRecordBatch batch = batches.get(0);
            return batch.baseOffset() == 0L &&
                    batch.baseSequence() == sequence &&
                    batch.producerId() == producerIdAndEpoch.producerId &&
                    batch.producerEpoch() == producerIdAndEpoch.epoch &&
                    batch.isTransactional() == isTransactional;
        };
    }
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