fix(shard-manager): fixed the ETCD integration test

Makefile

@@ -715,7 +715,7 @@ integration_tests_etcd:
 
 	$Q echo "Running integration tests with etcd"
 	$Q mkdir -p $(BUILD)/$(INTEG_TEST_DIR)
-	$Q (ETCD_TEST_DIRS=$$(find . -name "*_test.go" -exec grep -l "testflags.RequireEtcd" {} \; | xargs -n1 dirname | sort | uniq); \
+	$Q (ETCD_TEST_DIRS=$$(find . -name "*_test.go" -exec grep -l "testhelper.SetupStoreTestCluster\|testflags.RequireEtcd" {} \; | xargs -n1 dirname | sort | uniq); \
 		echo "Found etcd test directories:"; \
 		echo "$$ETCD_TEST_DIRS"; \
 		echo "Using ETCD_ENDPOINTS='$(ETCD_ENDPOINTS)'"; \

service/sharddistributor/store/etcd/executorstore/etcdstore_test.go

@@ -643,7 +643,7 @@ func TestDeleteShardStatsDeletesLargeBatches(t *testing.T) {
 	tc := testhelper.SetupStoreTestCluster(t)
 	executorStore := createStore(t, tc)
 
-	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 	defer cancel()
 
 	totalShardStats := deleteShardStatsBatchSize*2 + 7 // two batches + 7 extra (remainder)

service/sharddistributor/store/etcd/executorstore/shardcache/namespaceshardcache.go

@@ -143,7 +143,7 @@ func (n *namespaceShardToExecutor) refresh(ctx context.Context) error {
 	}
 	n.RUnlock()
 
-	n.pubSub.publish(n.executorState)
+	n.pubSub.publish(executorState)
 	return nil
 }
 

service/sharddistributor/store/etcd/executorstore/shardcache/namespaceshardcache_test.go

@@ -75,9 +75,11 @@ func TestNamespaceShardToExecutor_Lifecycle(t *testing.T) {
 	})
 
 	// Check the cache is populated
+	namespaceShardToExecutor.RLock()
 	_, ok := namespaceShardToExecutor.executorRevision["executor-1"]
 	assert.True(t, ok)
 	assert.Equal(t, "executor-1", namespaceShardToExecutor.shardToExecutor["shard-1"].ExecutorID)
+	namespaceShardToExecutor.RUnlock()
 
 	// Add executor-2 with shard-2 to trigger watch update
 	setupExecutorWithShards(t, testCluster, "test-ns", "executor-2", []string{"shard-2"}, map[string]string{
@@ -87,9 +89,11 @@ func TestNamespaceShardToExecutor_Lifecycle(t *testing.T) {
 	time.Sleep(100 * time.Millisecond)
 
 	// Check that executor-2 and shard-2 is in the cache
+	namespaceShardToExecutor.RLock()
 	_, ok = namespaceShardToExecutor.executorRevision["executor-2"]
 	assert.True(t, ok)
 	assert.Equal(t, "executor-2", namespaceShardToExecutor.shardToExecutor["shard-2"].ExecutorID)
+	namespaceShardToExecutor.RUnlock()
 
 	// Verify executor-2 owns shard-2 with correct metadata
 	verifyShardOwner(t, namespaceShardToExecutor, "shard-2", "executor-2", map[string]string{