chore: rocketmq broker add/remove

addons/rocketmq/scripts/broker-setup.sh

@@ -109,9 +109,57 @@ init_acl() {
     fi
 }
 
+sync_file() {
+    file_type=$1
+    if [ "$file_type" == "topics.json" ]; then
+        query_path="topicInfo"
+    elif [ "$file_type" == "subscriptionGroup.json" ]; then
+        query_path="subscriptionGroupInfo"
+    else
+        echo "Unknown file type: ${file_type}"
+    fi
+
+    if [ -f "${DATA_DIR}"/config/"${file_type}" ]; then
+        return
+    fi
+
+    brokers=$(eval echo "${ALL_BROKER_FQDN}" | tr '@' '\n')
+    if [ "$(echo "${brokers}" | wc -l)" -lt 2 ]; then
+        return
+    fi
+
+    for broker in ${brokers}; do
+        broker_name=$(echo "${broker}" | awk -F':' '{print $1}')
+        fqdns_str=$(echo "${broker}" | awk -F':' '{print $NF}')
+        if [ "$broker_name" == "${MY_COMP_NAME}" ]; then
+            continue
+        fi
+        fqdns=$(eval echo "${fqdns_str}" | tr ',' '\n')
+        for fqdn in ${fqdns}; do
+            if ! curl -X GET -H 'Content-Type: application/json' http://"${fqdn}:${ROCKETMQ_AGENT}"/"${query_path}" > /tmp/"${file_type}" 2>/dev/null; then
+                echo "Failed to fetch ${file_type} from ${fqdn}"
+                continue
+            else
+                echo "Successfully fetched ${file_type} from ${fqdn}"
+                break
+            fi
+        done
+
+        if grep -q "dataVersion" /tmp/"${file_type}"; then
+            if [ ! -d "${DATA_DIR}"/config ]; then
+                mkdir -p "${DATA_DIR}"/config
+            fi
+            mv /tmp/"${file_type}" "${DATA_DIR}"/config/"${file_type}"
+            return
+        fi
+    done
+}
+
 copy_log_config
 init_broker
 init_acl
+sync_file "topics.json"
+sync_file "subscriptionGroup.json"
 
 export NAMESRV_ADDR=${ALL_NAMESRV_FQDN//,/:${NAMESRV_PORT};}:${NAMESRV_PORT}
 if ! grep -q "NAMESRV_ADDR" ~/.bashrc; then

addons/rocketmq/scripts/remove-broker.sh

@@ -0,0 +1,27 @@
+#!/bin/bash
+
+export NAMESRV_ADDR=${ALL_NAMESRV_FQDN//,/:${NAMESRV_PORT};}:${NAMESRV_PORT}
+export PATH="$PATH:/opt/java/openjdk/bin"
+
+# 1. stop broker write
+/home/rocketmq/rocketmq-"${ROCKETMQ_VERSION}"/bin/mqadmin updateBrokerConfig -b "${MY_POD_IP}:${BROKER_PORT}" -k brokerPermission -v 4
+
+# 2. checking consume stats
+max_retries=3
+diff_left=true
+for ((i=1; i<=max_retries; i++)); do
+    stats=$(/home/rocketmq/rocketmq-"${ROCKETMQ_VERSION}"/bin/mqadmin brokerConsumeStats -b "${MY_POD_IP}:${BROKER_PORT}")
+    diff_total=$(echo "$stats" | grep "Diff Total" | awk -F': ' '{print $2}')
+    if [ "$diff_total" -eq 0 ]; then
+        echo "All messages have been consumed. Proceeding to remove the broker."
+        diff_left=false
+    else
+        echo "There are still unconsumed messages (Diff Total: $diff_total). Checking again in 1 seconds... ($i/$max_retries)"
+        sleep 1
+    fi
+done
+
+if [ "$diff_left" = true ]; then
+    echo "Error: There are still unconsumed messages after $max_retries checks. Aborting broker removal."
+    exit 1
+fi

addons/rocketmq/templates/cmpd-broker.yaml

@@ -91,6 +91,10 @@ spec:
       value: "{{ .Values.broker.port }}"
     - name: DLEDGER_PORT
       value: "{{ .Values.broker.dledgerPort }}"
+    - name: ROCKETMQ_AGENT
+      value: "{{ .Values.broker.agentPort }}"
+    - name: DATA_DIR
+      value: {{ .Values.broker.dataMountPath }}
     - name: MY_CLUSTER_NAME
       valueFrom:
         clusterVarRef:
@@ -163,8 +167,6 @@ spec:
           - mountPath: /home/rocketmq/rocketmq-4.9.6/conf
             name: conf
         env:
-          - name: DATA_DIR
-            value: {{ .Values.broker.dataMountPath }}
           - name: JMX_PORT
             value: "{{ .Values.broker.jxmPort }}"
           - name: MY_POD_NAME
@@ -200,6 +202,17 @@ spec:
         volumeMounts:
           - name: jmx-config
             mountPath: /etc/jmx-rocketmq
+      # agent is a light http server to get topic and subscriptionGroup info
+      - name: agent
+        imagePullPolicy: {{ default .Values.image.pullPolicy "IfNotPresent" }}
+        command:
+          - /bin/rocketmq_agent
+        volumeMounts:
+          - mountPath: {{ .Values.broker.dataMountPath }}
+            name: data
+        ports:
+          - name: agent
+            containerPort: {{ .Values.broker.agentPort }}
   systemAccounts:
     - name: rocketmq-admin
       initAccount: true

addons/rocketmq/templates/cmpd-nameserver.yaml

@@ -24,7 +24,7 @@ spec:
       namespace: {{ .Release.Namespace }}
   lifecycleActions:
     memberJoin:
-      timeoutSeconds: 3
+      timeoutSeconds: 30
       exec:
         container: rocketmq-namesrv
         image: {{ .Values.image.registry | default "docker.io" }}/{{ .Values.image.repository }}:{{ .Values.image.tag }}
@@ -34,7 +34,7 @@ spec:
           - |
             /scripts/update-nameserver-addrs.sh
     memberLeave:
-      timeoutSeconds: 3
+      timeoutSeconds: 30
       exec:
         command:
           - bash

addons/rocketmq/templates/cmpv-broker.yaml

@@ -13,6 +13,7 @@ spec:
       images:
         rocketmq-broker: {{ .Values.image.registry | default "docker.io" }}/{{ .Values.image.repository }}:{{ .Values.image.tag }}
         jmx-exporter: {{ .Values.image.registry | default "docker.io" }}/{{ .Values.image.jmxExporter.repository }}:{{ .Values.image.jmxExporter.tag }}
+        agent: {{ .Values.image.registry | default "docker.io" }}/{{ .Values.image.agent.repository }}:{{ .Values.image.agent.tag }}
 
   compatibilityRules:
     - releases:

addons/rocketmq/templates/scripts.yaml

@@ -11,6 +11,8 @@ data:
     {{- .Files.Get "scripts/get-role.sh" | nindent 4 }}
   util.sh: |-
     {{- .Files.Get "scripts/util.sh" | nindent 4 }}
+  remove-broker.sh: |-
+    {{- .Files.Get "scripts/remove-broker.sh" | nindent 4 }}
 
 ---
 apiVersion: v1

addons/rocketmq/templates/sharding-definition.yaml

@@ -13,4 +13,23 @@ spec:
     minShards: 1
     maxShards: 128
   provisionStrategy: Parallel
-  updateStrategy: Parallel
+  updateStrategy: Parallel
+  lifecycleActions:
+    shardRemove:
+      timeoutSeconds: 60
+      exec:
+        container: rocketmq-broker
+        targetPodSelector: Role
+        matchingKey: master
+        env:
+          - name: MY_POD_IP
+            valueFrom:
+              fieldRef:
+                apiVersion: v1
+                fieldPath: status.podIP
+        image: {{ .Values.image.registry | default "docker.io" }}/{{ .Values.image.repository }}:{{ .Values.image.tag }}
+        command:
+          - bash
+          - -c
+          - |
+            /scripts/remove-broker.sh

addons/rocketmq/values.yaml

@@ -24,6 +24,10 @@ image:
     repository: apecloud/jmx-exporter
     tag: 0.18.0-debian-11-r20
 
+  agent:
+    repository: apecloud/rocketmq-agent
+    tag: 1.0.0
+
 ## @param compDefinitionVersion for each ComponentDefinition resources name created by this chart, that can avoid name conflict
 componentDefinitionVersion:
   rocketMQ4: "4"
@@ -40,6 +44,7 @@ broker:
   port: 10911
   jxmPort: 5555
   dledgerPort: 40911
+  agentPort: 8999
 
 dashboard:
   dataPath: /configs