From 76ff0696fda318d637cf397bbe6dacc7e6bec25f Mon Sep 17 00:00:00 2001
From: dengziming <dengziming1993@gmail.com>
Date: Thu, 2 Jun 2022 11:25:30 +0800
Subject: [PATCH] MINOR: Use right enum value for broker registration change

---
 .../controller/ClusterControlManager.java     |  4 +-
 .../controller/ReplicationControlManager.java |  4 +-
 .../org/apache/kafka/image/ClusterDelta.java  |  4 +-
 .../BrokerRegistrationFencingChange.java      |  4 +-
 .../controller/ClusterControlManagerTest.java |  6 +--
 .../BrokerRegistrationFencingChangeTest.java  |  8 ++--
 ...trationInControlledShutdownChangeTest.java | 48 +++++++++++++++++++
 7 files changed, 63 insertions(+), 15 deletions(-)
 create mode 100644 metadata/src/test/java/org/apache/kafka/metadata/BrokerRegistrationInControlledShutdownChangeTest.java

diff --git a/metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java b/metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java
index 859dedda2200f..ad11e19ec2d74 100644
--- a/metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java
+++ b/metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java
@@ -422,7 +422,7 @@ public void replay(FenceBrokerRecord record) {
             record,
             record.id(),
             record.epoch(),
-            BrokerRegistrationFencingChange.UNFENCE.asBoolean(),
+            BrokerRegistrationFencingChange.FENCE.asBoolean(),
             BrokerRegistrationInControlledShutdownChange.NONE.asBoolean()
         );
     }
@@ -432,7 +432,7 @@ public void replay(UnfenceBrokerRecord record) {
             record,
             record.id(),
             record.epoch(),
-            BrokerRegistrationFencingChange.FENCE.asBoolean(),
+            BrokerRegistrationFencingChange.UNFENCE.asBoolean(),
             BrokerRegistrationInControlledShutdownChange.NONE.asBoolean()
         );
     }
diff --git a/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java b/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
index 8382cd9c16b7b..fc6ac38c31728 100644
--- a/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
+++ b/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java
@@ -1162,7 +1162,7 @@ void handleBrokerFenced(int brokerId, List<ApiMessageAndVersion> records) {
         if (featureControl.metadataVersion().isBrokerRegistrationChangeRecordSupported()) {
             records.add(new ApiMessageAndVersion(new BrokerRegistrationChangeRecord().
                     setBrokerId(brokerId).setBrokerEpoch(brokerRegistration.epoch()).
-                    setFenced(BrokerRegistrationFencingChange.UNFENCE.value()),
+                    setFenced(BrokerRegistrationFencingChange.FENCE.value()),
                     (short) 0));
         } else {
             records.add(new ApiMessageAndVersion(new FenceBrokerRecord().
@@ -1205,7 +1205,7 @@ void handleBrokerUnfenced(int brokerId, long brokerEpoch, List<ApiMessageAndVers
         if (featureControl.metadataVersion().isBrokerRegistrationChangeRecordSupported()) {
             records.add(new ApiMessageAndVersion(new BrokerRegistrationChangeRecord().
                 setBrokerId(brokerId).setBrokerEpoch(brokerEpoch).
-                setFenced(BrokerRegistrationFencingChange.FENCE.value()),
+                setFenced(BrokerRegistrationFencingChange.UNFENCE.value()),
                 (short) 0));
         } else {
             records.add(new ApiMessageAndVersion(new UnfenceBrokerRecord().setId(brokerId).
diff --git a/metadata/src/main/java/org/apache/kafka/image/ClusterDelta.java b/metadata/src/main/java/org/apache/kafka/image/ClusterDelta.java
index 110e68fb89a29..39d6fdb3d744c 100644
--- a/metadata/src/main/java/org/apache/kafka/image/ClusterDelta.java
+++ b/metadata/src/main/java/org/apache/kafka/image/ClusterDelta.java
@@ -94,7 +94,7 @@ private BrokerRegistration getBrokerOrThrow(int brokerId, long epoch, String act
     public void replay(FenceBrokerRecord record) {
         BrokerRegistration curRegistration = getBrokerOrThrow(record.id(), record.epoch(), "fence");
         changedBrokers.put(record.id(), Optional.of(curRegistration.cloneWith(
-            BrokerRegistrationFencingChange.UNFENCE.asBoolean(),
+            BrokerRegistrationFencingChange.FENCE.asBoolean(),
             Optional.empty()
         )));
     }
@@ -102,7 +102,7 @@ public void replay(FenceBrokerRecord record) {
     public void replay(UnfenceBrokerRecord record) {
         BrokerRegistration curRegistration = getBrokerOrThrow(record.id(), record.epoch(), "unfence");
         changedBrokers.put(record.id(), Optional.of(curRegistration.cloneWith(
-            BrokerRegistrationFencingChange.FENCE.asBoolean(),
+            BrokerRegistrationFencingChange.UNFENCE.asBoolean(),
             Optional.empty()
         )));
     }
diff --git a/metadata/src/main/java/org/apache/kafka/metadata/BrokerRegistrationFencingChange.java b/metadata/src/main/java/org/apache/kafka/metadata/BrokerRegistrationFencingChange.java
index 1c2729355fc94..ab0bfc6a81f84 100644
--- a/metadata/src/main/java/org/apache/kafka/metadata/BrokerRegistrationFencingChange.java
+++ b/metadata/src/main/java/org/apache/kafka/metadata/BrokerRegistrationFencingChange.java
@@ -25,9 +25,9 @@
 
 
 public enum BrokerRegistrationFencingChange {
-    FENCE(-1, Optional.of(false)),
+    FENCE(1, Optional.of(true)),
     NONE(0, Optional.empty()),
-    UNFENCE(1, Optional.of(true));
+    UNFENCE(-1, Optional.of(false));
 
     private final byte value;
 
diff --git a/metadata/src/test/java/org/apache/kafka/controller/ClusterControlManagerTest.java b/metadata/src/test/java/org/apache/kafka/controller/ClusterControlManagerTest.java
index c0b163180e59f..97d6c88377277 100644
--- a/metadata/src/test/java/org/apache/kafka/controller/ClusterControlManagerTest.java
+++ b/metadata/src/test/java/org/apache/kafka/controller/ClusterControlManagerTest.java
@@ -111,7 +111,7 @@ public void testReplay(MetadataVersion metadataVersion) {
             clusterControl.replay(unfenceBrokerRecord);
         } else {
             BrokerRegistrationChangeRecord changeRecord =
-                    new BrokerRegistrationChangeRecord().setBrokerId(1).setBrokerEpoch(100).setFenced((byte) -1);
+                    new BrokerRegistrationChangeRecord().setBrokerId(1).setBrokerEpoch(100).setFenced(BrokerRegistrationFencingChange.UNFENCE.value());
             clusterControl.replay(changeRecord);
         }
         assertFalse(clusterControl.unfenced(0));
@@ -123,7 +123,7 @@ public void testReplay(MetadataVersion metadataVersion) {
             clusterControl.replay(fenceBrokerRecord);
         } else {
             BrokerRegistrationChangeRecord changeRecord =
-                    new BrokerRegistrationChangeRecord().setBrokerId(1).setBrokerEpoch(100).setFenced((byte) 1);
+                    new BrokerRegistrationChangeRecord().setBrokerId(1).setBrokerEpoch(100).setFenced(BrokerRegistrationFencingChange.FENCE.value());
             clusterControl.replay(changeRecord);
         }
         assertFalse(clusterControl.unfenced(0));
@@ -234,7 +234,7 @@ public void testReplayBrokerRegistrationChangeRecord() {
         registrationChangeRecord = new BrokerRegistrationChangeRecord()
             .setBrokerId(0)
             .setBrokerEpoch(100)
-            .setFenced(BrokerRegistrationFencingChange.FENCE.value());
+            .setFenced(BrokerRegistrationFencingChange.UNFENCE.value());
         clusterControl.replay(registrationChangeRecord);
 
         assertTrue(clusterControl.unfenced(0));
diff --git a/metadata/src/test/java/org/apache/kafka/metadata/BrokerRegistrationFencingChangeTest.java b/metadata/src/test/java/org/apache/kafka/metadata/BrokerRegistrationFencingChangeTest.java
index 9266192f24042..8f48923c99373 100644
--- a/metadata/src/test/java/org/apache/kafka/metadata/BrokerRegistrationFencingChangeTest.java
+++ b/metadata/src/test/java/org/apache/kafka/metadata/BrokerRegistrationFencingChangeTest.java
@@ -29,16 +29,16 @@
 public class BrokerRegistrationFencingChangeTest {
     @Test
     public void testValues() {
-        assertEquals((byte) -1, BrokerRegistrationFencingChange.FENCE.value());
+        assertEquals((byte) 1, BrokerRegistrationFencingChange.FENCE.value());
         assertEquals((byte) 0, BrokerRegistrationFencingChange.NONE.value());
-        assertEquals((byte) 1, BrokerRegistrationFencingChange.UNFENCE.value());
+        assertEquals((byte) -1, BrokerRegistrationFencingChange.UNFENCE.value());
     }
 
     @Test
     public void testAsBoolean() {
-        assertEquals(Optional.of(false), BrokerRegistrationFencingChange.FENCE.asBoolean());
+        assertEquals(Optional.of(true), BrokerRegistrationFencingChange.FENCE.asBoolean());
         assertEquals(Optional.empty(), BrokerRegistrationFencingChange.NONE.asBoolean());
-        assertEquals(Optional.of(true), BrokerRegistrationFencingChange.UNFENCE.asBoolean());
+        assertEquals(Optional.of(false), BrokerRegistrationFencingChange.UNFENCE.asBoolean());
     }
 
     @Test
diff --git a/metadata/src/test/java/org/apache/kafka/metadata/BrokerRegistrationInControlledShutdownChangeTest.java b/metadata/src/test/java/org/apache/kafka/metadata/BrokerRegistrationInControlledShutdownChangeTest.java
new file mode 100644
index 0000000000000..7f6b69031c40c
--- /dev/null
+++ b/metadata/src/test/java/org/apache/kafka/metadata/BrokerRegistrationInControlledShutdownChangeTest.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.metadata;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import java.util.Optional;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+@Timeout(40)
+public class BrokerRegistrationInControlledShutdownChangeTest {
+
+    @Test
+    public void testValues() {
+        assertEquals((byte) 0, BrokerRegistrationInControlledShutdownChange.NONE.value());
+        assertEquals((byte) 1, BrokerRegistrationInControlledShutdownChange.IN_CONTROLLED_SHUTDOWN.value());
+    }
+
+    @Test
+    public void testAsBoolean() {
+        assertEquals(Optional.empty(), BrokerRegistrationInControlledShutdownChange.NONE.asBoolean());
+        assertEquals(Optional.of(true), BrokerRegistrationInControlledShutdownChange.IN_CONTROLLED_SHUTDOWN.asBoolean());
+    }
+
+    @Test
+    public void testValueRoundTrip() {
+        for (BrokerRegistrationInControlledShutdownChange change : BrokerRegistrationInControlledShutdownChange.values()) {
+            assertEquals(Optional.of(change), BrokerRegistrationInControlledShutdownChange.fromValue(change.value()));
+        }
+    }
+}