private void verifyErrorMessage(ProduceResponse response, String expectedMessage) throws Exception {
        Future<RecordMetadata> future = appendToAccumulator(tp0, 0L, "key", "value");
        sender.runOnce(); // connect
        sender.runOnce(); // send produce request
        client.respond(response);
        sender.runOnce();
        sender.runOnce();
        ExecutionException e1 = assertThrows(ExecutionException.class, () -> future.get(5, TimeUnit.SECONDS));
        assertEquals(InvalidRequestException.class, e1.getCause().getClass());
        assertEquals(expectedMessage, e1.getCause().getMessage());
    }
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