feat: controller handles planned termination

common/src/main/java/com/automq/rocketmq/common/config/BrokerConfig.java

@@ -66,6 +66,8 @@ public class BrokerConfig implements ControllerConfig {
 
     private final DatabaseConfig db;
 
+    private boolean terminating;
+
     public BrokerConfig() {
         this.metrics = new MetricsConfig();
         this.proxy = new ProxyConfig();
@@ -206,6 +208,11 @@ public String advertiseAddress() {
         return advertiseAddress;
     }
 
+    @Override
+    public boolean goingAway() {
+        return terminating;
+    }
+
     public void setName(String name) {
         this.name = name;
     }

common/src/main/java/com/automq/rocketmq/common/config/ControllerConfig.java

@@ -76,4 +76,6 @@ default int workloadTolerance() {
 
     String dbPassword();
 
+    boolean goingAway();
+
 }

controller/src/main/java/com/automq/rocketmq/controller/metadata/database/DefaultMetadataStore.java

@@ -294,7 +294,7 @@ public void heartbeat() {
 
         try {
             String target = leaderAddress();
-            controllerClient.heartbeat(target, config.nodeId(), config.epoch(), false)
+            controllerClient.heartbeat(target, config.nodeId(), config.epoch(), config.goingAway())
                 .whenComplete((r, e) -> {
                     if (null != e) {
                         LOGGER.error("Failed to maintain heartbeat to {}", target);

controller/src/main/java/com/automq/rocketmq/controller/tasks/LeaseTask.java

@@ -53,6 +53,12 @@ public void process() throws ControllerException {
                         // Someone must have won the leader election campaign.
                         return;
                     }
+
+                    if (metadataStore.config().goingAway()) {
+                        LOGGER.info("Current node is terminating thus should stop renewing its leadership lease");
+                        return;
+                    }
+
                     Calendar calendar = Calendar.getInstance();
                     calendar.add(Calendar.SECOND, metadataStore.config().leaseLifeSpanInSecs());
                     update.setExpirationTime(calendar.getTime());
@@ -70,6 +76,11 @@ public void process() throws ControllerException {
                         lease.getNodeId(), lease.getEpoch(), lease.getExpirationTime());
                 }
             } else {
+                if (metadataStore.config().goingAway()) {
+                    LOGGER.debug("Current node is terminating and will not campaign for leadership");
+                    return;
+                }
+
                 // Perform leader election campaign
                 Lease update = leaseMapper.currentWithWriteLock();
                 if (lease.getEpoch() == update.getEpoch() && lease.getNodeId() == update.getNodeId()) {

controller/src/test/java/com/automq/rocketmq/controller/metadata/DatabaseTestBase.java

@@ -70,6 +70,7 @@ public DatabaseTestBase() {
         Mockito.when(config.deletedGroupLingersInSecs()).thenCallRealMethod();
         Mockito.when(config.balanceWorkloadIntervalInSecs()).thenCallRealMethod();
         Mockito.when(config.recycleS3IntervalInSecs()).thenCallRealMethod();
+        Mockito.when(config.goingAway()).thenReturn(false);
         gson = new GsonBuilder()
             .registerTypeAdapter(SubStream.class, new SubStreamSerializer())
             .registerTypeAdapter(SubStream.class, new SubStreamDeserializer())

distribution/conf/broker.yaml

@@ -6,11 +6,11 @@ innerSecretKey: ""
 s3Stream:
   s3WALPath: "/tmp/s3rocketmq/wal"
   s3Endpoint: "http://minio.hellocorp.test"
-  s3Bucket: "lzh"
+  s3Bucket: "lizhanhui"
   s3Region: "us-east-1"
   s3ForcePathStyle: true
-  s3AccessKey: "mcFoJPLvxNmRq497dhK0"
-  s3SecretKey: "UU7PE0VOo3fRtqYEBeIIQzOSGFT3GctuiLoZg2hY"
+  s3AccessKey: "kicaSWtNBPf7XCCsdW8Z"
+  s3SecretKey: "dgWxasHBEQBBXxAxBpFmp4VWIoQ0XyDU3gGdFnaT"
 db:
   url: "jdbc:mysql://mysql-server:3306/metadata"
   userName: "root"