MINOR: Use right enum value for broker registration change

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()
         );
     }

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).

metadata/src/main/java/org/apache/kafka/image/ClusterDelta.java

@@ -94,15 +94,15 @@ 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()
         )));
     }
 
     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()
         )));
     }

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;
 

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));

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

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()));
+        }
+    }
+}