AndrewJSchofield commented on code in PR #23965:
URL: https://github.com/apache/kafka/pull/23965#discussion_r4457883761
##########
server/src/main/java/org/apache/kafka/server/partition/DefaultAlterPartitionManager.java:
##########
@@ -56,18 +56,20 @@ public class DefaultAlterPartitionManager implements
AlterPartitionManager {
private final Scheduler scheduler;
private final int brokerId;
private final Supplier<Long> brokerEpochSupplier;
+ private final Supplier<Short> requestHeaderVersion;
// Used to allow only one pending ISR update per partition (visible for
testing)
final ConcurrentHashMap<TopicIdPartition, AlterPartitionItem>
unsentIsrUpdates = new ConcurrentHashMap<>();
// Used to allow only one in-flight request at a time
private final AtomicBoolean inflightRequest = new AtomicBoolean(false);
- public DefaultAlterPartitionManager(NodeToControllerChannelManager
controllerChannelManager, Scheduler scheduler, int brokerId, Supplier<Long>
brokerEpochSupplier) {
+ public DefaultAlterPartitionManager(NodeToControllerChannelManager
controllerChannelManager, Scheduler scheduler, int brokerId, Supplier<Long>
brokerEpochSupplier, Supplier<Short> requestHeaderVersion) {
this.controllerChannelManager = controllerChannelManager;
this.scheduler = scheduler;
this.brokerId = brokerId;
this.brokerEpochSupplier = brokerEpochSupplier;
+ this.requestHeaderVersion = requestHeaderVersion;
Review Comment:
This really ought to be `requestHeaderVersionSupplier` I think. This comment
applies to the entire call stack for this piece of information.
##########
server/src/main/java/org/apache/kafka/server/partition/DefaultAlterPartitionManager.java:
##########
@@ -217,7 +220,7 @@ private AlterPartitionRequest.Builder buildRequest(
.setLeaderRecoveryState(item.leaderAndIsr().leaderRecoveryState().value()));
}
- return new AlterPartitionRequest.Builder(message);
+ return new AlterPartitionRequest.Builder(message,
requestHeaderVersion.get());
Review Comment:
The supplied probably ought to be a bit more resilient.
##########
transaction-coordinator/src/main/java/org/apache/kafka/coordinator/transaction/RPCProducerIdManager.java:
##########
@@ -66,12 +67,14 @@ public class RPCProducerIdManager implements
ProducerIdManager {
public RPCProducerIdManager(int brokerId,
Time time,
Supplier<Long> brokerEpochSupplier,
- NodeToControllerChannelManager
controllerChannel
+ NodeToControllerChannelManager
controllerChannel,
+ Supplier<Short> requestHeaderVersion
Review Comment:
This is a supplier, not the version.
##########
transaction-coordinator/src/main/java/org/apache/kafka/coordinator/transaction/RPCProducerIdManager.java:
##########
@@ -136,7 +139,7 @@ protected void sendRequest() {
var message = new AllocateProducerIdsRequestData()
.setBrokerEpoch(brokerEpochSupplier.get())
.setBrokerId(brokerId);
- var request = new AllocateProducerIdsRequest.Builder(message);
+ var request = new AllocateProducerIdsRequest.Builder(message,
requestHeaderVersion.get());
Review Comment:
I think there's the possibility that the supplier throws and that would be
very inconvenient. Maybe worth copying the fallback code which was used in
share group request header versions derived from MV.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
