From 1f9577c8161d35cca7bd31c5d51022798c3f9e58 Mon Sep 17 00:00:00 2001 From: panxiao81 Date: Fri, 25 Sep 2026 18:41:47 +0000 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BD=BF=E7=94=A8=E5=90=88=E6=B3=95?= =?UTF-8?q?=E7=9A=84=20JetStream=20consumer=20=E5=90=8D=E7=A7=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/assignmentqueue/jetstream.go | 8 ++++++-- internal/assignmentqueue/jetstream_test.go | 2 +- 2 files changed, 7 insertions(+), 3 deletions(-) diff --git a/internal/assignmentqueue/jetstream.go b/internal/assignmentqueue/jetstream.go index 80de9d5..67ec0d2 100644 --- a/internal/assignmentqueue/jetstream.go +++ b/internal/assignmentqueue/jetstream.go @@ -186,9 +186,13 @@ func OpenConsumer(ctx context.Context, manager consumerManager, stream, subjectB return nil, err } key := placement.Key() + // JetStream consumer names may not contain dots even though subjects do. + // Keep the placement subject hierarchical while using a stable, legal name + // for the durable cursor. + consumerName := strings.ReplaceAll(key, ".", "-") consumer, err := manager.CreateOrUpdateConsumer(ctx, stream, jetstream.ConsumerConfig{ - Name: key, - Durable: key, + Name: consumerName, + Durable: consumerName, FilterSubject: strings.TrimSuffix(subjectBase, ".") + "." + key, AckPolicy: jetstream.AckExplicitPolicy, AckWait: 5 * time.Minute, diff --git a/internal/assignmentqueue/jetstream_test.go b/internal/assignmentqueue/jetstream_test.go index 9c1490f..74a6857 100644 --- a/internal/assignmentqueue/jetstream_test.go +++ b/internal/assignmentqueue/jetstream_test.go @@ -215,7 +215,7 @@ func TestOpenConsumerUsesIndependentDurablePerPlacement(t *testing.T) { if _, err := OpenConsumer(context.Background(), manager, "CI_RUNNER", "ci.assignment", taskassignment.KubernetesContainer, 4); err != nil { t.Fatal(err) } - if manager.config.Durable != "container.kubernetes" || manager.config.FilterSubject != "ci.assignment.container.kubernetes" || manager.config.AckPolicy != jetstream.AckExplicitPolicy || manager.config.MaxAckPending != 4 || manager.config.MaxDeliver != 1000 { + if manager.config.Name != "container-kubernetes" || manager.config.Durable != "container-kubernetes" || manager.config.FilterSubject != "ci.assignment.container.kubernetes" || manager.config.AckPolicy != jetstream.AckExplicitPolicy || manager.config.MaxAckPending != 4 || manager.config.MaxDeliver != 1000 { t.Fatalf("config = %#v", manager.config) } } -- 2.54.0