jolshan commented on code in PR #13607:
URL: https://github.com/apache/kafka/pull/13607#discussion_r1180864129
##########
clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java:
##########
@@ -222,6 +223,27 @@ public void clearPartitionRecords() {
partitionSizes();
data = null;
}
+
+ public static void validateProducerIds(short version, ProduceRequestData
data) {
+ if (version >= 3) {
+ long producerId = -1;
+ for (ProduceRequestData.TopicProduceData topicData :
data.topicData()) {
+ for (ProduceRequestData.PartitionProduceData partitionData :
topicData.partitionData()) {
+ BaseRecords baseRecords = partitionData.records();
+ if (baseRecords instanceof Records) {
+ Records records = (Records) baseRecords;
+ for (RecordBatch batch : records.batches()) {
+ if (producerId == -1 && batch.hasProducerId())
Review Comment:
Based on some of the ProduceRequest tests I found, we allow for some batches
with and some without.
See `testMixedIdempotentData`
--
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]
