Divij Vaidya

Test failures are unrelated. @showuon this is ready for your review. Test failures: ``` [2022-09-06T16:49:33.427Z] org.apache.kafka.controller.QuorumControllerTest.testEarlyControllerResults() failed, log available in /home/jenkins/workspace/Kafka_kafka-pr_PR-12590/metadata/build/reports/testOutput/org.apache.kafka.controller.QuorumControllerTest.testEarlyControllerResults().test.stdout [2022-09-06T16:49:33.427Z] [2022-09-06T16:49:33.427Z] QuorumControllerTest > testEarlyControllerResults() FAILED [2022-09-06T16:49:33.427Z] org.apache.kafka.server.fault.FaultHandlerException: fatalFaultHandler:...

@dajac since you are working on the consumer client protocol, perhaps you may also be interested in taking a look at this PR?

> does the fetch request to close the session fetches any records No, because the fetch request's field for topic partitions is set to empty at `sessionHandler.newBuilder().build()` (line 1963 at...

The failing tests pass locally and are unrelated to this change. ``` Build / JDK 8 and Scala 2.12 / testReturnRecordsDuringRebalance() – org.apache.kafka.clients.consumer.KafkaConsumerTest Build / JDK 17 and Scala 2.13...

@dajac this is ready for your review, whenever you get a chance

@dajac please take a look when you get a chance!

@kirktrue @dajac please take a look when you get a chance!

Test failures are unrelated: ``` Build / JDK 8 and Scala 2.12 / testFailureToFenceEpoch(String).quorum=kraft – kafka.api.TransactionsTest Build / JDK 11 and Scala 2.13 / shouldWorkWithUncleanShutdownWipeOutStateStore[exactly_once_v2] – org.apache.kafka.streams.integration.EOSUncleanShutdownIntegrationTest ```

@showuon please take a look whenever you get a chance.

@hachikuji this is ready for your review. The test failures are unrelated: ``` Build / JDK 8 and Scala 2.12 / testTransactionAfterProducerIdExpires(String).quorum=zk – kafka.api.TransactionsExpirationTest Build / JDK 8 and Scala...