Controller should return the correct number of sub-responses when an error occurs

[Chillax-0v0]

related #603

server.log

[2023-12-27 03:31:44,875] ERROR Response size: 0 not match request size: 1 (kafka.log.stream.s3.network.ControllerRequestSender)
[2023-12-27 03:31:44,882] ERROR [Stream id=14886 epoch=2] trim fail (com.automq.stream.s3.S3Stream)
java.lang.RuntimeException: Response size not match request size
	at kafka.log.stream.s3.network.ControllerRequestSender$RequestAccumulator$1.onSuccess(ControllerRequestSender.java:195)
	at kafka.log.stream.s3.network.ControllerRequestSender$2.onComplete(ControllerRequestSender.java:135)
	at kafka.server.BrokerToControllerRequestThread.handleResponse(BrokerToControllerChannelManager.scala:413)
	at kafka.server.BrokerToControllerRequestThread.$anonfun$generateRequests$1(BrokerToControllerChannelManager.scala:376)
	at org.apache.kafka.clients.ClientResponse.onComplete(ClientResponse.java:109)
	at org.apache.kafka.clients.NetworkClient.completeResponses(NetworkClient.java:583)
	at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:575)
	at kafka.common.InterBrokerSendThread.pollOnce(InterBrokerSendThread.scala:74)
	at kafka.server.BrokerToControllerRequestThread.doWork(BrokerToControllerChannelManager.scala:421)
	at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:96)
[2023-12-27 03:31:44,877] ERROR Trim stream[14886] (new offset = 1435003) failed, retry later (kafka.log.streamaspect.AlwaysSuccessClient)
java.lang.RuntimeException: Response size not match request size
	at kafka.log.stream.s3.network.ControllerRequestSender$RequestAccumulator$1.onSuccess(ControllerRequestSender.java:195)
	at kafka.log.stream.s3.network.ControllerRequestSender$2.onComplete(ControllerRequestSender.java:135)
	at kafka.server.BrokerToControllerRequestThread.handleResponse(BrokerToControllerChannelManager.scala:413)
	at kafka.server.BrokerToControllerRequestThread.$anonfun$generateRequests$1(BrokerToControllerChannelManager.scala:376)
	at org.apache.kafka.clients.ClientResponse.onComplete(ClientResponse.java:109)
	at org.apache.kafka.clients.NetworkClient.completeResponses(NetworkClient.java:583)
	at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:575)
	at kafka.common.InterBrokerSendThread.pollOnce(InterBrokerSendThread.scala:74)
	at kafka.server.BrokerToControllerRequestThread.doWork(BrokerToControllerChannelManager.scala:421)
	at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:96)

[superhx]

[superhx]

[2024-06-07 10:37:46,050] WARN [Controller 0] openStream: failed with unknown server exception NotLeaderException at epoch 2 in 311 us. Renouncing leadership and reverting to the last committed offset 16711. (org.apache.kafka.controller.QuorumController)