[kafka] branch trunk updated: MINOR; Fix partition change record noop check (#12073)

[jsancio]

This is an automated email from the ASF dual-hosted git repository.

jsancio pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/trunk by this push:
     new 4380eae7ce MINOR; Fix partition change record noop check (#12073)
4380eae7ce is described below

commit 4380eae7ceb840dd93fee8ec90cd89a72bad7a3f
Author: José Armando García Sancio <jsan...@users.noreply.github.com>
AuthorDate: Thu Apr 21 09:05:46 2022 -0700

    MINOR; Fix partition change record noop check (#12073)
    
    When LeaderRecoveryState was added to the PartitionChangeRecord, the
    check for being a noop was not updated. This commit fixes that and
    improves the associated test to avoid this oversight in the future.
    
    Reviewers: Colin Patrick McCabe <cmcc...@apache.org>
---
 .../kafka/common/protocol/types/TaggedFields.java       |  7 +++++++
 .../apache/kafka/controller/PartitionChangeBuilder.java |  1 +
 .../kafka/controller/PartitionChangeBuilderTest.java    | 17 +++++++++++++++++
 3 files changed, 25 insertions(+)

diff --git 
a/clients/src/main/java/org/apache/kafka/common/protocol/types/TaggedFields.java
 
b/clients/src/main/java/org/apache/kafka/common/protocol/types/TaggedFields.java
index 4e1ab0d4d5..129f80c90b 100644
--- 
a/clients/src/main/java/org/apache/kafka/common/protocol/types/TaggedFields.java
+++ 
b/clients/src/main/java/org/apache/kafka/common/protocol/types/TaggedFields.java
@@ -178,4 +178,11 @@ public class TaggedFields extends DocumentedType {
     public String documentation() {
         return "Represents a series of tagged fields.";
     }
+
+    /**
+     * The number of tagged fields
+     */
+    public int numFields() {
+        return this.fields.size();
+    }
 }
diff --git 
a/metadata/src/main/java/org/apache/kafka/controller/PartitionChangeBuilder.java
 
b/metadata/src/main/java/org/apache/kafka/controller/PartitionChangeBuilder.java
index 1bbef97c11..907994d107 100644
--- 
a/metadata/src/main/java/org/apache/kafka/controller/PartitionChangeBuilder.java
+++ 
b/metadata/src/main/java/org/apache/kafka/controller/PartitionChangeBuilder.java
@@ -48,6 +48,7 @@ public class PartitionChangeBuilder {
         if (record.replicas() != null) return false;
         if (record.removingReplicas() != null) return false;
         if (record.addingReplicas() != null) return false;
+        if (record.leaderRecoveryState() != LeaderRecoveryState.NO_CHANGE) 
return false;
         return true;
     }
 
diff --git 
a/metadata/src/test/java/org/apache/kafka/controller/PartitionChangeBuilderTest.java
 
b/metadata/src/test/java/org/apache/kafka/controller/PartitionChangeBuilderTest.java
index 06c54322ec..fedfa8c0a5 100644
--- 
a/metadata/src/test/java/org/apache/kafka/controller/PartitionChangeBuilderTest.java
+++ 
b/metadata/src/test/java/org/apache/kafka/controller/PartitionChangeBuilderTest.java
@@ -19,6 +19,7 @@ package org.apache.kafka.controller;
 
 import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.metadata.PartitionChangeRecord;
+import org.apache.kafka.common.protocol.types.TaggedFields;
 import org.apache.kafka.controller.PartitionChangeBuilder.ElectionResult;
 import org.apache.kafka.metadata.LeaderRecoveryState;
 import org.apache.kafka.metadata.PartitionRegistration;
@@ -47,6 +48,16 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
 public class PartitionChangeBuilderTest {
     @Test
     public void testChangeRecordIsNoOp() {
+        /* If the next few checks fail please update them based on the latest 
schema and make sure
+         * to update changeRecordIsNoOp to take into account the new schema or 
tagged fields.
+         */
+        // Check that the supported versions haven't changed
+        assertEquals(0, PartitionChangeRecord.HIGHEST_SUPPORTED_VERSION);
+        assertEquals(0, PartitionChangeRecord.LOWEST_SUPPORTED_VERSION);
+        // For the latest version check that the number of tagged fields 
hasn't changed
+        TaggedFields taggedFields = (TaggedFields) 
PartitionChangeRecord.SCHEMA_0.get(2).def.type;
+        assertEquals(6, taggedFields.numFields());
+
         assertTrue(changeRecordIsNoOp(new PartitionChangeRecord()));
         assertFalse(changeRecordIsNoOp(new 
PartitionChangeRecord().setLeader(1)));
         assertFalse(changeRecordIsNoOp(new PartitionChangeRecord().
@@ -55,6 +66,12 @@ public class PartitionChangeBuilderTest {
             setRemovingReplicas(Arrays.asList(1))));
         assertFalse(changeRecordIsNoOp(new PartitionChangeRecord().
             setAddingReplicas(Arrays.asList(4))));
+        assertFalse(
+            changeRecordIsNoOp(
+                new PartitionChangeRecord()
+                  
.setLeaderRecoveryState(LeaderRecoveryState.RECOVERED.value())
+            )
+        );
     }
 
     private final static PartitionRegistration FOO = new PartitionRegistration(