RockteMQ-AI commented on code in PR #56:
URL: https://github.com/apache/rocketmq-operator/pull/56#discussion_r3902494704
##########
pkg/share/share.go:
##########
@@ -18,19 +18,45 @@
// Package share defines some variables shared by different packages
package share
+import "sync"
+
var (
Review Comment:
`ShareItem` is a value type (struct), not a pointer. `LoadOrStore` and
`Load` return a copy; mutations to fields like `actual.NameServersStr` are
local until `Store` is explicitly called. When broker and nameservice
controllers reconcile concurrently for the same key, they each get independent
copies, mutate them, and the last `Store` wins — silently dropping the other's
changes (e.g., `IsNameServersStrInitialized` could be overwritten back to
`false`). Use `*ShareItem` (pointer) or a mutex-guarded struct to ensure atomic
read-modify-write.
##########
pkg/controller/broker/broker_controller.go:
##########
@@ -131,31 +131,43 @@ func (r *ReconcileBroker) Reconcile(request
reconcile.Request) (reconcile.Result
return reconcile.Result{}, err
}
+ actualKey := broker.Namespace + "-" + broker.Spec.RocketMQName
+ actual, _ := share.GetInstance().LoadOrStore(actualKey,
share.ShareItem{})
+ defer func() {
+ reqLogger.Info("Broker actualKey:" + actualKey + "
actual.NameServerStr:" + actual.NameServersStr +
+ " actual.BrokerClusterName:" + actual.BrokerClusterName
+ " IsNameServersStrInitialized:" +
strconv.FormatBool(actual.IsNameServersStrInitialized))
+ share.GetInstance().Store(actualKey, actual)
+ }()
+
+ var groupNum int
if broker.Status.Size == 0 {
- share.GroupNum = broker.Spec.Size
+ groupNum = broker.Spec.Size
} else {
- share.GroupNum = broker.Status.Size
+ groupNum = broker.Status.Size
}
Review Comment:
Infinite blocking `for` loop with `time.Sleep` has no timeout, context
cancellation, or requeue. If the NameServer never becomes ready, this
permanently blocks a controller worker goroutine, starving all other reconcile
requests. Replace with `return reconcile.Result{RequeueAfter: ...}, nil` so the
work is re-enqueued without blocking.
##########
deploy/crds/rocketmq_v1alpha1_nameservice_crd.yaml:
##########
@@ -73,6 +76,7 @@ spec:
- storageMode
- hostPath
- volumeClaimTemplates
+ - rocketMQName
Review Comment:
Same backward-compatibility break as the Broker CRD: `rocketMQName` is added
to `required`, which will reject any existing NameService CR that lacks the
field upon upgrade.
##########
deploy/crds/rocketmq_v1alpha1_broker_crd.yaml:
##########
@@ -93,6 +96,7 @@ spec:
- volumes
- volumeClaimTemplates
- scalePodName
+ - rocketMQName
Review Comment:
`rocketMQName` is added to the `required` list. Any existing Broker CR that
does not include this field will fail validation after CRD upgrade, breaking
backward compatibility. Either provide a default value via an admission webhook
/ mutating webhook, remove it from `required`, or use a conversion/migration
step for existing resources.
##########
deploy/clusterrole_binding.yaml:
##########
@@ -0,0 +1,27 @@
+# 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.
+
+kind: ClusterRoleBinding
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+ name: rocketmq-operator
+subjects:
+ - kind: ServiceAccount
+ name: rocketmq-operator
Review Comment:
The ServiceAccount namespace is hardcoded to `default`. If the operator is
deployed to any other namespace, the binding will reference a non-existent
ServiceAccount and RBAC will deny all API calls. Make this configurable (e.g.,
via Kustomize or Helm templating).
##########
pkg/controller/broker/broker_controller.go:
##########
@@ -189,8 +201,8 @@ func (r *ReconcileBroker) Reconcile(request
reconcile.Request) (reconcile.Result
// Check for name server scaling
if broker.Spec.AllowRestart {
// The following code will restart all brokers to update
NAMESRV_ADDR env
- if share.IsNameServersStrUpdated {
- for brokerGroupIndex := 0; brokerGroupIndex <
broker.Spec.Size; brokerGroupIndex++ {
+ if actual.IsNameServersStrUpdated {
Review Comment:
Indentation regression: the `for` loop inside the `if
actual.IsNameServersStrUpdated` block has an extra tab, misaligning it from the
enclosing `if`. This is likely a merge artifact and should be corrected to
maintain readability.
##########
pkg/controller/broker/broker_controller.go:
##########
@@ -454,9 +475,13 @@ func (r *ReconcileBroker) getBrokerStatefulSet(broker
*rocketmqv1alpha1.Broker,
}
func getENV(broker *rocketmqv1alpha1.Broker, replicaIndex int,
brokerGroupIndex int) []corev1.EnvVar {
+
Review Comment:
`getENV` calls `LoadOrStore(actualKey, share.ShareItem{})` which will insert
an empty `ShareItem` (all zero values) if the key does not yet exist. If the
broker reconciles before the nameservice controller has stored real values,
`NameServersStr` will be empty and broker pods will be created with an invalid
`NAMESRV_ADDR`. Consider using `Load` instead and returning an error / requeue
when the key is absent.
##########
pkg/controller/nameservice/nameservice_controller.go:
##########
@@ -174,7 +174,23 @@ func (r *ReconcileNameService)
updateNameServiceStatus(instance *rocketmqv1alpha
}
hostIps := getNameServers(podList.Items)
+ if len(hostIps) != int(instance.Spec.Size) || len(hostIps) == 0 {
Review Comment:
Early return when `len(hostIps) != instance.Spec.Size` happens before the
deferred `Store`, so no `ShareItem` is created for this key. This is correct
behavior, but if the nameservice never reaches full size (e.g., a node
shortage), brokers waiting on `IsNameServersStrInitialized` will block
indefinitely (see broker_controller.go infinite loop issue). Consider setting a
partial-readiness indicator or adding a timeout.
##########
deploy/cluster_role.yaml:
##########
@@ -0,0 +1,72 @@
+# 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.
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRole
+metadata:
+ creationTimestamp: null
+ name: rocketmq-operator
+rules:
+ - apiGroups:
+ - ""
+ resources:
+ - pods
+ - services
+ - endpoints
+ - persistentvolumeclaims
+ - events
+ - configmaps
+ - secrets
+ - pods/exec
+ verbs:
+ - '*'
+ - apiGroups:
+ - ""
+ resources:
+ - namespaces
+ verbs:
+ - get
+ - apiGroups:
+ - apps
+ resources:
+ - deployments
+ - daemonsets
+ - replicasets
+ - statefulsets
+ verbs:
+ - '*'
+ - apiGroups:
+ - monitoring.coreos.com
+ resources:
+ - servicemonitors
+ verbs:
+ - get
+ - create
+ - apiGroups:
+ - apps
+ resourceNames:
+ - rocketmq-operator
+ resources:
+ - deployments/finalizers
+ verbs:
+ - update
+ - apiGroups:
+ - rocketmq.apache.org
+ resources:
Review Comment:
`pods/exec` is listed under the `rocketmq.apache.org` API group, which is
incorrect — `pods/exec` belongs to the core (`""`) API group and is already
listed there (line 30). The duplicate under the custom API group is meaningless
and should be removed.
##########
pkg/apis/rocketmq/v1alpha1/topictransfer_types.go:
##########
@@ -37,6 +37,8 @@ type TopicTransferSpec struct {
SourceCluster string `json:"sourceCluster,omitempty"`
// The cluster where the topic will be transferred to
TargetCluster string `json:"targetCluster,omitempty"`
+ // // RocketMQ Name, the broker and nameserver in the same cluster
must be filled with the same name
Review Comment:
Malformed comment: `// // RocketMQ Name...` contains a stray tab and
duplicate `//` prefix, likely a copy-paste artifact. Clean up to `// RocketMQ
Name, the broker and nameserver in the same cluster must be filled with the
same name`.
##########
pkg/controller/topictransfer/topictransfer_controller.go:
##########
@@ -127,7 +127,9 @@ func (r *ReconcileTopicTransfer) Reconcile(request
reconcile.Request) (reconcile
targetCluster := topicTransfer.Spec.TargetCluster
sourceCluster := topicTransfer.Spec.SourceCluster
- nameServer := strings.Split(share.NameServersStr, ";")[0]
+ actualKey := topicTransfer.Namespace + "-" +
topicTransfer.Spec.RocketMQName
Review Comment:
`LoadOrStore` with an empty `ShareItem` means if the TopicTransfer
reconciles before any NameService has populated the map, an empty entry is
persisted. `strings.Split("", ";")[0]` returns `""` which is caught by the
length check, but the stale empty entry remains in the map. Use `Load` and
requeue if the key is not found.
##########
pkg/controller/broker/broker_controller.go:
##########
@@ -276,6 +295,17 @@ func (r *ReconcileBroker) Reconcile(request
reconcile.Request) (reconcile.Result
cmd = []string{"/bin/bash", "-c", MakeConfigDirCommand + " && "
+ ChmodDirCommand + " && " + topicsCommand + " && " + subscriptionGroupCommand}
}
+ // Update status.Nodes if needed
Review Comment:
The new `len(podNames) == 0` guard returns early with a requeue, but the
subsequent status update at line 302 adds a `len(broker.Status.Nodes) != 0`
guard that prevents clearing `Status.Nodes` when all pods are gone. This
asymmetry means the status can never reflect a fully-scaled-down state; if that
is intentional (safety), document it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
