Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
1f9577c816
|
||
|
|
6ea78d4704 |
@@ -186,9 +186,13 @@ func OpenConsumer(ctx context.Context, manager consumerManager, stream, subjectB
|
|||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
key := placement.Key()
|
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{
|
consumer, err := manager.CreateOrUpdateConsumer(ctx, stream, jetstream.ConsumerConfig{
|
||||||
Name: key,
|
Name: consumerName,
|
||||||
Durable: key,
|
Durable: consumerName,
|
||||||
FilterSubject: strings.TrimSuffix(subjectBase, ".") + "." + key,
|
FilterSubject: strings.TrimSuffix(subjectBase, ".") + "." + key,
|
||||||
AckPolicy: jetstream.AckExplicitPolicy,
|
AckPolicy: jetstream.AckExplicitPolicy,
|
||||||
AckWait: 5 * time.Minute,
|
AckWait: 5 * time.Minute,
|
||||||
|
|||||||
@@ -215,7 +215,7 @@ func TestOpenConsumerUsesIndependentDurablePerPlacement(t *testing.T) {
|
|||||||
if _, err := OpenConsumer(context.Background(), manager, "CI_RUNNER", "ci.assignment", taskassignment.KubernetesContainer, 4); err != nil {
|
if _, err := OpenConsumer(context.Background(), manager, "CI_RUNNER", "ci.assignment", taskassignment.KubernetesContainer, 4); err != nil {
|
||||||
t.Fatal(err)
|
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)
|
t.Fatalf("config = %#v", manager.config)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user