Copilot commented on code in PR #992: URL: https://github.com/apache/dubbo-go-pixiu/pull/992#discussion_r3479808699
########## pkg/client/mq/kafka_facade_test.go: ########## @@ -0,0 +1,149 @@ +/* + * 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 mq + +import ( + "sync" + "testing" + "time" +) + +// TestKafkaConsumerFacadeConsumerManagerInitialized verifies that +// the consumerManager map is initialized during construction, +// preventing nil map panics during Subscribe operations. +func TestKafkaConsumerFacadeConsumerManagerInitialized(t *testing.T) { + facade := &KafkaConsumerFacade{ + consumerManager: make(map[string]func()), + done: make(chan struct{}), + mu: sync.RWMutex{}, + } + + testKey := "test-topic-test-group" + + // This should NOT panic due to nil map write + defer func() { + if r := recover(); r != nil { + t.Fatalf("Subscribe panicked with nil map: %v", r) + } + }() + + facade.mu.Lock() + facade.consumerManager[testKey] = func() {} + facade.mu.Unlock() + + facade.mu.RLock() + _, exists := facade.consumerManager[testKey] + facade.mu.RUnlock() + + if !exists { + t.Error("Expected key to exist in consumerManager") + } +} + +// TestKafkaConsumerFacadeConcurrentAccess verifies that concurrent +// access to consumerManager is safe due to mutex protection. +func TestKafkaConsumerFacadeConcurrentAccess(t *testing.T) { + facade := &KafkaConsumerFacade{ + consumerManager: make(map[string]func()), + done: make(chan struct{}), + mu: sync.RWMutex{}, + } + + done := make(chan bool) + + // Writer goroutine + go func() { + for i := 0; i < 100; i++ { + facade.mu.Lock() + facade.consumerManager[GetConsumerManagerKey([]string{"topic"}, "group")] = func() {} + facade.mu.Unlock() + } + done <- true + }() + + // Reader goroutine + go func() { + for i := 0; i < 100; i++ { + facade.mu.RLock() + _ = len(facade.consumerManager) + facade.mu.RUnlock() + } + done <- true + }() + + <-done + <-done +} Review Comment: `TestKafkaConsumerFacadeConcurrentAccess` can hang indefinitely if either goroutine fails to signal on `done` (e.g., due to an unexpected panic). Adding a bounded wait (timeout) makes the test suite more reliable. ########## pkg/client/mq/kafka_facade.go: ########## @@ -186,10 +195,17 @@ func (f *KafkaConsumerFacade) checkConsumerIsAlive(ctx context.Context, key stri } if lastCheck != http.StatusOK { - f.consumerManager[key]() - delete(f.consumerManager, key) + f.mu.Lock() + if cancel, ok := f.consumerManager[key]; ok { + cancel() + delete(f.consumerManager, key) + } + f.mu.Unlock() } + case <-ctx.Done(): + ticker.Stop() + return Review Comment: The new `case <-ctx.Done()` path stops the ticker and returns, but does not remove `key` from `consumerManager`. If the subscription context is canceled externally (e.g., request timeout), this leaves a stale entry behind. ########## pkg/client/mq/kafka_facade.go: ########## @@ -81,7 +87,9 @@ func (f *KafkaConsumerFacade) Subscribe(ctx context.Context, opts ...Option) err cOpt.ApplyOpts(opts...) c, cancel := context.WithCancel(ctx) key := GetConsumerManagerKey(cOpt.TopicList, cOpt.ConsumerGroup) + f.mu.Lock() f.consumerManager[key] = cancel + f.mu.Unlock() f.wg.Add(2) go f.consumeLoop(ctx, cOpt.TopicList, &consumerGroupHandler{cOpt.ConsumeUrl, f.httpClient}) go f.checkConsumerIsAlive(c, key, cOpt.CheckUrl) Review Comment: `Subscribe` increments the WaitGroup by 2, but only `checkConsumerIsAlive` calls `wg.Done()`. `consumeLoop` never calls `wg.Done()` and also blocks on `<-f.done` outside a `select`, so `Stop()` can hang forever waiting for a goroutine that never decrements the WaitGroup (and may never start consuming). ########## pkg/client/mq/kafka_facade.go: ########## @@ -81,7 +87,9 @@ func (f *KafkaConsumerFacade) Subscribe(ctx context.Context, opts ...Option) err cOpt.ApplyOpts(opts...) c, cancel := context.WithCancel(ctx) key := GetConsumerManagerKey(cOpt.TopicList, cOpt.ConsumerGroup) + f.mu.Lock() f.consumerManager[key] = cancel + f.mu.Unlock() f.wg.Add(2) go f.consumeLoop(ctx, cOpt.TopicList, &consumerGroupHandler{cOpt.ConsumeUrl, f.httpClient}) go f.checkConsumerIsAlive(c, key, cOpt.CheckUrl) Review Comment: `Subscribe` stores a cancel func derived from `c := context.WithCancel(ctx)`, but `consumeLoop` is started with the *parent* `ctx`. When `checkConsumerIsAlive` calls the stored cancel, it will not stop `consumeLoop`, leaving the consumer running even after being deemed unhealthy. ########## pkg/client/mq/kafka_facade_test.go: ########## @@ -0,0 +1,149 @@ +/* + * 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 mq + +import ( + "sync" + "testing" + "time" +) + +// TestKafkaConsumerFacadeConsumerManagerInitialized verifies that +// the consumerManager map is initialized during construction, +// preventing nil map panics during Subscribe operations. +func TestKafkaConsumerFacadeConsumerManagerInitialized(t *testing.T) { + facade := &KafkaConsumerFacade{ + consumerManager: make(map[string]func()), + done: make(chan struct{}), + mu: sync.RWMutex{}, + } + + testKey := "test-topic-test-group" + + // This should NOT panic due to nil map write + defer func() { + if r := recover(); r != nil { + t.Fatalf("Subscribe panicked with nil map: %v", r) + } + }() + + facade.mu.Lock() + facade.consumerManager[testKey] = func() {} + facade.mu.Unlock() + + facade.mu.RLock() + _, exists := facade.consumerManager[testKey] + facade.mu.RUnlock() + + if !exists { + t.Error("Expected key to exist in consumerManager") + } +} + +// TestKafkaConsumerFacadeConcurrentAccess verifies that concurrent +// access to consumerManager is safe due to mutex protection. +func TestKafkaConsumerFacadeConcurrentAccess(t *testing.T) { + facade := &KafkaConsumerFacade{ + consumerManager: make(map[string]func()), + done: make(chan struct{}), + mu: sync.RWMutex{}, + } + + done := make(chan bool) + + // Writer goroutine + go func() { + for i := 0; i < 100; i++ { + facade.mu.Lock() + facade.consumerManager[GetConsumerManagerKey([]string{"topic"}, "group")] = func() {} + facade.mu.Unlock() + } + done <- true + }() + + // Reader goroutine + go func() { + for i := 0; i < 100; i++ { + facade.mu.RLock() + _ = len(facade.consumerManager) + facade.mu.RUnlock() + } + done <- true + }() + + <-done + <-done +} + +// TestGetConsumerManagerKey verifies the key generation function +func TestGetConsumerManagerKey(t *testing.T) { + topics := []string{"topic1", "topic2"} + group := "test-group" + + key := GetConsumerManagerKey(topics, group) + + expectedKey := GetConsumerManagerKey(topics, group) + if key != expectedKey { + t.Error("Key should be deterministic") + } + + differentTopics := []string{"topic3", "topic4"} + differentKey := GetConsumerManagerKey(differentTopics, group) + if key == differentKey { + t.Error("Different topics should produce different keys") + } + + differentGroup := "other-group" + differentGroupKey := GetConsumerManagerKey(topics, differentGroup) + if key == differentGroupKey { + t.Error("Different groups should produce different keys") + } +} + +// TestNewKafkaConsumerFacadeConstructorInitializesMap verifies that +// the consumerManager map is initialized during construction via NewKafkaConsumerFacade. +func TestNewKafkaConsumerFacadeConstructorInitializesMap(t *testing.T) { + if testing.Short() { + t.Skip("Skipping test that requires Kafka broker in short mode") + } + + config := KafkaConsumerConfig{ + Brokers: []string{"localhost:9092"}, + ClientID: "test-client", + ProtocolVersion: "2.8.0", + Metadata: Metadata{ Review Comment: This test depends on a real Kafka broker (`localhost:9092`), which makes CI runs flaky/slow and the result environment-dependent. Prefer skipping by default unless an explicit broker address is provided via env var (or move this into a separate integration test file with a build tag). ########## pkg/client/mq/kafka_facade.go: ########## @@ -160,6 +168,7 @@ func (f *KafkaConsumerFacade) checkConsumerIsAlive(ctx context.Context, key stri select { case <-f.done: ticker.Stop() + return Review Comment: On shutdown via `f.done`, `checkConsumerIsAlive` returns without removing its `consumerManager` entry. That can leave stale cancel funcs in the map (and is inconsistent with the cleanup performed when a consumer is detected unhealthy). ########## pkg/client/mq/kafka_facade_test.go: ########## @@ -0,0 +1,149 @@ +/* + * 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 mq + +import ( + "sync" + "testing" + "time" +) Review Comment: `TestNewKafkaConsumerFacadeConstructorInitializesMap` currently hard-codes `localhost:9092`. If this test is kept as an integration test gated by an environment variable, it needs `os` for `os.Getenv`. -- 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] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
