Skip to content

Commit f37c88e

Browse files
EldertGrootenboerEldertGrootenboer
authored
[Service Bus] Add topic SQLFilterCount and CorrelationFilterCount runtime properties (#27323)
* [Service Bus] Add topic SQLFilterCount and CorrelationFilterCount runtime properties Add read-only SQLFilterCount and CorrelationFilterCount fields to TopicRuntimeProperties, reporting the total number of SQL and correlation filters across a topic's subscriptions. These are served by Service Bus management API version 2024-05, so the admin client now requests 2024-05. - Expose SQLFilterCount/CorrelationFilterCount (int32, 0 when absent) - Parse the SqlFilterCount/CorrelationFilterCount ATOM elements - Request api-version 2024-05 for admin operations - Add unit and live tests - Bump module to v1.11.0-beta.1 (new feature) Part of a cross-language effort; reference implementation is Azure/azure-sdk-for-net#61559. * Elide embedded TopicRuntimeProperties field in filter-count test (staticcheck QF1008) * Address PR review: add ATOM-XML deserialization tests for topic filter counts + document zero-when-absent
1 parent 69c6c7e commit f37c88e

10 files changed

Lines changed: 361 additions & 6 deletions

‎sdk/messaging/azservicebus/CHANGELOG.md‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,20 +1,23 @@
11
# Release History
22

3-
## 1.10.1-beta.1 (Unreleased)
3+
## 1.11.0-beta.1 (Unreleased)
44

55
### Features Added
66

77
- Added `Client.NewListSessionsForQueuePager()` and `Client.NewListSessionsForSubscriptionPager()` to list the IDs of sessions in session-enabled queues and subscriptions. By default they list sessions that have active messages, as well as sessions that have session state set but no active messages; set `SessionStateUpdatedAfter` to instead list sessions whose session state was updated after a given time. (PR#26688)
8+
- Added `SQLFilterCount` and `CorrelationFilterCount` to `TopicRuntimeProperties`, exposing the total number of SQL filters and correlation filters across all of a topic's subscriptions, including the default rule each subscription is created with. The service reports these at api-version 2024-05 or later. (PR#27323)
89

910
### Breaking Changes
1011

1112
### Bugs Fixed
1213

14+
- Setting `APIVersion` in the administration client's `ClientOptions` now takes effect. Setting it previously made every administration call fail with "this client doesn't support overriding its API version". (PR#27323)
1315
- Management operations (PeekMessages, ScheduleMessages, CancelScheduledMessages, and others) now send a `server-timeout` that expires one second before the caller's context, so the broker answers first and the caller gets a service-side timeout instead of `context deadline exceeded`. When the context has no deadline, each attempt asks the broker to answer within 60 seconds, where it was previously given no bound at all. The client still waits only on its context, so set one to cap the call itself. (#26421)
1416
- Read `com.microsoft:max-message-batch-size` vendor property from the AMQP sender link to correctly limit batch size on Premium large-message entities, where `max-message-size` can be up to 100 MB but the batch limit is 1 MB.
1517

1618
### Other Changes
1719

20+
- Every administration operation now runs against service api-version `2024-05`, where it previously ran against `2021-05`. Responses follow the 2024-05 contract, which adds fields to some entity descriptions, and the topic filter counts above require this version. Set `APIVersion` on the `azcore.ClientOptions` embedded in the client's `ClientOptions` to pin a different version. (PR#27323)
1821
- Cleaned up accumulated `golangci-lint` findings in `azservicebus` (deprecated
1922
`runtime.WithHTTPHeader` calls switched to `policy.WithHTTPHeader`, duplicate
2023
`log` package imports consolidated under the `azlog` alias, unchecked `Close`

‎sdk/messaging/azservicebus/admin/admin_client.go‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,17 @@ type RetryOptions = exported.RetryOptions
2828

2929
// ClientOptions allows you to set optional configuration for `Client`.
3030
type ClientOptions struct {
31+
// Administration operations run against service api-version 2024-05. Set APIVersion to pin a
32+
// different one, for example "2021-05" or "2017-04". APIVersion is promoted from the embedded
33+
// type, so a composite literal has to name that type:
34+
//
35+
// opts := &admin.ClientOptions{
36+
// ClientOptions: azcore.ClientOptions{APIVersion: "2021-05"},
37+
// }
38+
//
39+
// The value is sent as-is, so it has to be a version the Service Bus management endpoint
40+
// supports. A version earlier than 2024-05 omits TopicRuntimeProperties.SQLFilterCount and
41+
// CorrelationFilterCount, which then read 0.
3142
azcore.ClientOptions
3243
}
3344

‎sdk/messaging/azservicebus/admin/admin_client_apiversion_test.go‎

Lines changed: 80 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,80 @@
1+
// Copyright (c) Microsoft Corporation. All rights reserved.
2+
// Licensed under the MIT License.
3+
4+
package admin
5+
6+
import (
7+
"context"
8+
"io"
9+
"net/http"
10+
"strings"
11+
"testing"
12+
"time"
13+
14+
"github.com/Azure/azure-sdk-for-go/sdk/azcore"
15+
"github.com/Azure/azure-sdk-for-go/sdk/azcore/policy"
16+
"github.com/stretchr/testify/require"
17+
)
18+
19+
const apiVersionTopicXML = `<entry xmlns="http://www.w3.org/2005/Atom"><title>my-topic</title><content type="application/xml"><TopicDescription xmlns="http://schemas.microsoft.com/netservices/2010/10/servicebus/connect"><SubscriptionCount>1</SubscriptionCount><CountDetails><ScheduledMessageCount>0</ScheduledMessageCount></CountDetails><CreatedAt>2026-01-01T00:00:00Z</CreatedAt><UpdatedAt>2026-01-01T00:00:00Z</UpdatedAt><AccessedAt>2026-01-01T00:00:00Z</AccessedAt></TopicDescription></content></entry>`
20+
21+
// captureAPIVersionPolicy records the api-version the client is about to send and answers the
22+
// request itself, so nothing reaches the network.
23+
type captureAPIVersionPolicy struct {
24+
apiVersions []string
25+
}
26+
27+
func (p *captureAPIVersionPolicy) Do(req *policy.Request) (*http.Response, error) {
28+
// The whole slice rather than Get(), so a second api-version parameter is visible here
29+
// rather than hidden behind the first one.
30+
p.apiVersions = req.Raw().URL.Query()["api-version"]
31+
32+
return &http.Response{
33+
StatusCode: http.StatusOK,
34+
Body: io.NopCloser(strings.NewReader(apiVersionTopicXML)),
35+
Request: req.Raw(),
36+
}, nil
37+
}
38+
39+
// fakeAPIVersionCredential is never invoked: captureAPIVersionPolicy answers the request at the
40+
// per-call stage, before the per-retry auth policy runs. It exists because NewClient requires a
41+
// credential.
42+
type fakeAPIVersionCredential struct{}
43+
44+
func (fakeAPIVersionCredential) GetToken(ctx context.Context, options policy.TokenRequestOptions) (azcore.AccessToken, error) {
45+
return azcore.AccessToken{Token: "fake-token", ExpiresOn: time.Now().Add(time.Hour)}, nil
46+
}
47+
48+
// TestClientAPIVersion covers the promise the ClientOptions doc comment makes, at the public
49+
// boundary a customer touches. The atom-level TestEntityManagerAPIVersion covers the pipeline
50+
// wiring; this covers the options plumbing through NewClient that carries a caller's APIVersion
51+
// down to it.
52+
func TestClientAPIVersion(t *testing.T) {
53+
testData := []struct {
54+
name string
55+
override string
56+
expected string
57+
}{
58+
{name: "default", override: "", expected: "2024-05"},
59+
{name: "overridden", override: "2021-05", expected: "2021-05"},
60+
}
61+
62+
for _, td := range testData {
63+
t.Run(td.name, func(t *testing.T) {
64+
capture := &captureAPIVersionPolicy{}
65+
66+
client, err := NewClient("fake.servicebus.windows.net", fakeAPIVersionCredential{}, &ClientOptions{
67+
ClientOptions: azcore.ClientOptions{
68+
APIVersion: td.override,
69+
PerCallPolicies: []policy.Policy{capture},
70+
},
71+
})
72+
require.NoError(t, err)
73+
74+
_, err = client.GetTopicRuntimeProperties(context.Background(), "my-topic", nil)
75+
require.NoError(t, err)
76+
77+
require.Equal(t, []string{td.expected}, capture.apiVersions)
78+
})
79+
}
80+
}

‎sdk/messaging/azservicebus/admin/admin_client_topic.go‎

Lines changed: 18 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -77,6 +77,19 @@ type TopicRuntimeProperties struct {
7777
// SubscriptionCount is the number of subscriptions to the topic.
7878
SubscriptionCount int32
7979

80+
// SQLFilterCount is the total number of SQL filters across all subscriptions of the topic,
81+
// including the default rule each subscription is created with. The service reports this only at
82+
// api-version 2024-05 or later, so the field is 0 when the response omits it, which happens when
83+
// ClientOptions.APIVersion pins an earlier version or the namespace's service build predates the
84+
// field.
85+
SQLFilterCount int32
86+
87+
// CorrelationFilterCount is the total number of correlation filters across all subscriptions of
88+
// the topic. The service reports this only at api-version 2024-05 or later, so the field is 0
89+
// when the response omits it, which happens when ClientOptions.APIVersion pins an earlier version
90+
// or the namespace's service build predates the field.
91+
CorrelationFilterCount int32
92+
8093
// ScheduledMessageCount is the number of messages that are scheduled to be entopicd.
8194
ScheduledMessageCount int32
8295
}
@@ -414,9 +427,11 @@ func newTopicRuntimePropertiesItem(env *atom.TopicEnvelope) (*TopicRuntimeProper
414427
}
415428

416429
props := &TopicRuntimeProperties{
417-
SizeInBytes: int64OrZero(desc.SizeInBytes),
418-
ScheduledMessageCount: int32OrZero(desc.CountDetails.ScheduledMessageCount),
419-
SubscriptionCount: int32OrZero(desc.SubscriptionCount),
430+
SizeInBytes: int64OrZero(desc.SizeInBytes),
431+
ScheduledMessageCount: int32OrZero(desc.CountDetails.ScheduledMessageCount),
432+
SubscriptionCount: int32OrZero(desc.SubscriptionCount),
433+
SQLFilterCount: int32OrZero(desc.SQLFilterCount),
434+
CorrelationFilterCount: int32OrZero(desc.CorrelationFilterCount),
420435
}
421436

422437
var err error

‎sdk/messaging/azservicebus/admin/admin_client_topic_filtercount_live_test.go‎

Lines changed: 63 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,63 @@
1+
// Copyright (c) Microsoft Corporation. All rights reserved.
2+
// Licensed under the MIT License.
3+
4+
package admin
5+
6+
import (
7+
"context"
8+
"fmt"
9+
"testing"
10+
"time"
11+
12+
"github.com/Azure/azure-sdk-for-go/sdk/azcore/to"
13+
"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/test"
14+
"github.com/stretchr/testify/require"
15+
)
16+
17+
// TestAdminClient_TopicFilterCounts_Live exercises the topic filter-count feature
18+
// end to end: it creates a topic with a subscription, reads the baseline counts, adds
19+
// SQL and correlation filter rules, then asserts the topic-level counts moved by one
20+
// each. The counts are served at api-version 2024-05, which the admin client sends by
21+
// default; a namespace whose service build does not report them skips the test.
22+
func TestAdminClient_TopicFilterCounts_Live(t *testing.T) {
23+
adminClient := newAdminClientForTest(t, &test.NewClientOptions[ClientOptions]{})
24+
25+
topicName := fmt.Sprintf("topic-fc-%X", time.Now().UnixNano())
26+
_, err := adminClient.CreateTopic(context.Background(), topicName, nil)
27+
require.NoError(t, err)
28+
defer deleteTopic(t, adminClient, topicName)
29+
30+
subscriptionName := "sub1"
31+
_, err = adminClient.CreateSubscription(context.Background(), topicName, subscriptionName, nil)
32+
require.NoError(t, err)
33+
34+
before, err := adminClient.GetTopicRuntimeProperties(context.Background(), topicName, nil)
35+
require.NoError(t, err)
36+
37+
// A new subscription carries a default $Default rule, which is a SQL TrueFilter, and the service
38+
// counts it - a subscription with no explicit rules reads SQLFilterCount 1. So a zero here means
39+
// the namespace does not report the counts at all.
40+
if before.SQLFilterCount == 0 {
41+
t.Skipf("namespace does not report topic filter counts (sql=%d corr=%d)",
42+
before.SQLFilterCount, before.CorrelationFilterCount)
43+
}
44+
45+
_, err = adminClient.CreateRule(context.Background(), topicName, subscriptionName, &CreateRuleOptions{
46+
Name: to.Ptr("sqlrule"),
47+
Filter: &SQLFilter{Expression: "1=1"},
48+
})
49+
require.NoError(t, err)
50+
_, err = adminClient.CreateRule(context.Background(), topicName, subscriptionName, &CreateRuleOptions{
51+
Name: to.Ptr("corrrule"),
52+
Filter: &CorrelationFilter{CorrelationID: to.Ptr("abc")},
53+
})
54+
require.NoError(t, err)
55+
56+
after, err := adminClient.GetTopicRuntimeProperties(context.Background(), topicName, nil)
57+
require.NoError(t, err)
58+
59+
// Asserting the delta proves the counts track rule creation, without assuming how the
60+
// service counts the default rule.
61+
require.Equal(t, before.SQLFilterCount+1, after.SQLFilterCount)
62+
require.Equal(t, before.CorrelationFilterCount+1, after.CorrelationFilterCount)
63+
}

‎sdk/messaging/azservicebus/admin/admin_client_topic_filtercount_test.go‎

Lines changed: 96 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,96 @@
1+
// Copyright (c) Microsoft Corporation. All rights reserved.
2+
// Licensed under the MIT License.
3+
4+
package admin
5+
6+
import (
7+
"encoding/xml"
8+
"testing"
9+
10+
"github.com/Azure/azure-sdk-for-go/sdk/messaging/azservicebus/internal/atom"
11+
"github.com/stretchr/testify/require"
12+
)
13+
14+
func i32Ptr(v int32) *int32 { return &v }
15+
16+
func TestNewTopicRuntimePropertiesItem_FilterCounts(t *testing.T) {
17+
env := &atom.TopicEnvelope{
18+
Entry: &atom.Entry{Title: "my-topic"},
19+
Content: &atom.TopicContent{
20+
TopicDescription: atom.TopicDescription{
21+
SubscriptionCount: i32Ptr(2),
22+
SQLFilterCount: i32Ptr(7),
23+
CorrelationFilterCount: i32Ptr(9),
24+
CountDetails: &atom.CountDetails{ScheduledMessageCount: i32Ptr(1)},
25+
CreatedAt: "2026-01-01T00:00:00Z",
26+
UpdatedAt: "2026-01-01T00:00:00Z",
27+
AccessedAt: "2026-01-01T00:00:00Z",
28+
},
29+
},
30+
}
31+
32+
item, err := newTopicRuntimePropertiesItem(env)
33+
require.NoError(t, err)
34+
require.Equal(t, int32(2), item.SubscriptionCount)
35+
require.Equal(t, int32(7), item.SQLFilterCount)
36+
require.Equal(t, int32(9), item.CorrelationFilterCount)
37+
}
38+
39+
func TestNewTopicRuntimePropertiesItem_FilterCountsDefaultZero(t *testing.T) {
40+
// A namespace that does not report the counts omits the SqlFilterCount/CorrelationFilterCount
41+
// elements; the counts must default to zero.
42+
env := &atom.TopicEnvelope{
43+
Entry: &atom.Entry{Title: "my-topic"},
44+
Content: &atom.TopicContent{
45+
TopicDescription: atom.TopicDescription{
46+
SubscriptionCount: i32Ptr(1),
47+
CountDetails: &atom.CountDetails{ScheduledMessageCount: i32Ptr(0)},
48+
CreatedAt: "2026-01-01T00:00:00Z",
49+
UpdatedAt: "2026-01-01T00:00:00Z",
50+
AccessedAt: "2026-01-01T00:00:00Z",
51+
},
52+
},
53+
}
54+
55+
item, err := newTopicRuntimePropertiesItem(env)
56+
require.NoError(t, err)
57+
require.Equal(t, int32(1), item.SubscriptionCount)
58+
require.Zero(t, item.SQLFilterCount)
59+
require.Zero(t, item.CorrelationFilterCount)
60+
}
61+
62+
func TestNewTopicRuntimePropertiesItem_FilterCountsFromXML(t *testing.T) {
63+
// Unmarshal a real ATOM topic response through the same xml tags the client uses, so a typo in
64+
// the SqlFilterCount/CorrelationFilterCount xml tag is caught in CI. The other tests populate the
65+
// struct directly and would not detect a wrong tag.
66+
//
67+
// Captured 2026-08-13 from a live GET .../my-topic?api-version=2024-05 against a Standard
68+
// namespace, on a topic with one subscription carrying $Default, one SQL rule and one
69+
// correlation rule. Host and topic name scrubbed; every element is as the service emitted it.
70+
const topicXML = `<entry xmlns="http://www.w3.org/2005/Atom"><id>https://CONTOSO.servicebus.windows.net/my-topic?api-version=2024-05</id><title type="text">my-topic</title><published>2026-08-14T00:46:28Z</published><updated>2026-08-14T00:46:28Z</updated><author><name>CONTOSO</name></author><link rel="self" href="https://CONTOSO.servicebus.windows.net/my-topic?api-version=2024-05"/><content type="application/xml"><TopicDescription xmlns="http://schemas.microsoft.com/netservices/2010/10/servicebus/connect" xmlns:i="http://www.w3.org/2001/XMLSchema-instance"><DefaultMessageTimeToLive>P10675199DT2H48M5.4775807S</DefaultMessageTimeToLive><MaxSizeInMegabytes>1024</MaxSizeInMegabytes><RequiresDuplicateDetection>false</RequiresDuplicateDetection><DuplicateDetectionHistoryTimeWindow>PT10M</DuplicateDetectionHistoryTimeWindow><EnableBatchedOperations>true</EnableBatchedOperations><SizeInBytes>0</SizeInBytes><FilteringMessagesBeforePublishing>false</FilteringMessagesBeforePublishing><IsAnonymousAccessible>false</IsAnonymousAccessible><AuthorizationRules></AuthorizationRules><Status>Active</Status><CreatedAt>2026-08-14T00:46:28.4314346Z</CreatedAt><UpdatedAt>2026-08-14T00:46:28.4314346Z</UpdatedAt><AccessedAt>2026-08-14T00:46:32.9628806Z</AccessedAt><SupportOrdering>true</SupportOrdering><CountDetails xmlns:d2p1="http://schemas.microsoft.com/netservices/2011/06/servicebus"><d2p1:ActiveMessageCount>0</d2p1:ActiveMessageCount><d2p1:DeadLetterMessageCount>0</d2p1:DeadLetterMessageCount><d2p1:ScheduledMessageCount>0</d2p1:ScheduledMessageCount><d2p1:TransferMessageCount>0</d2p1:TransferMessageCount><d2p1:TransferDeadLetterMessageCount>0</d2p1:TransferDeadLetterMessageCount></CountDetails><SubscriptionCount>1</SubscriptionCount><AutoDeleteOnIdle>P10675199DT2H48M5.4775807S</AutoDeleteOnIdle><EnablePartitioning>false</EnablePartitioning><EntityAvailabilityStatus>Available</EntityAvailabilityStatus><EnableSubscriptionPartitioning>false</EnableSubscriptionPartitioning><EnableExpress>false</EnableExpress><MaxMessageSizeInKilobytes>256</MaxMessageSizeInKilobytes><SqlFilterCount>2</SqlFilterCount><CorrelationFilterCount>1</CorrelationFilterCount></TopicDescription></content></entry>`
71+
72+
var env *atom.TopicEnvelope
73+
require.NoError(t, xml.Unmarshal([]byte(topicXML), &env))
74+
75+
item, err := newTopicRuntimePropertiesItem(env)
76+
require.NoError(t, err)
77+
require.Equal(t, int32(1), item.SubscriptionCount)
78+
// The subscription's $Default rule is a SQL TrueFilter and the service counts it, so one
79+
// explicit SQL rule reads as 2.
80+
require.Equal(t, int32(2), item.SQLFilterCount)
81+
require.Equal(t, int32(1), item.CorrelationFilterCount)
82+
}
83+
84+
func TestNewTopicRuntimePropertiesItem_FilterCountsAbsentFromXML(t *testing.T) {
85+
// An older api-version omits the filter-count elements entirely; the counts must default to 0.
86+
const topicXML = `<entry xmlns="http://www.w3.org/2005/Atom"><title>my-topic</title><content type="application/xml"><TopicDescription xmlns="http://schemas.microsoft.com/netservices/2010/10/servicebus/connect"><SubscriptionCount>1</SubscriptionCount><CountDetails><ScheduledMessageCount>0</ScheduledMessageCount></CountDetails><CreatedAt>2026-01-01T00:00:00Z</CreatedAt><UpdatedAt>2026-01-01T00:00:00Z</UpdatedAt><AccessedAt>2026-01-01T00:00:00Z</AccessedAt></TopicDescription></content></entry>`
87+
88+
var env *atom.TopicEnvelope
89+
require.NoError(t, xml.Unmarshal([]byte(topicXML), &env))
90+
91+
item, err := newTopicRuntimePropertiesItem(env)
92+
require.NoError(t, err)
93+
require.Equal(t, int32(1), item.SubscriptionCount)
94+
require.Zero(t, item.SQLFilterCount)
95+
require.Zero(t, item.CorrelationFilterCount)
96+
}

‎sdk/messaging/azservicebus/internal/atom/entity_manager.go‎

Lines changed: 18 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,15 @@ const (
2626
serviceBusSchema = "http://schemas.microsoft.com/netservices/2010/10/servicebus/connect"
2727
atomSchema = "http://www.w3.org/2005/Atom"
2828
applicationXML = "application/xml"
29+
30+
// apiVersionQueryParam is the query parameter that carries the Service Bus
31+
// management API version.
32+
apiVersionQueryParam = "api-version"
33+
34+
// defaultAPIVersion is the management API version sent when the caller has not
35+
// set azcore.ClientOptions.APIVersion. The admin.ClientOptions doc comment names
36+
// this version, so the two move together.
37+
defaultAPIVersion = "2024-05"
2938
)
3039

3140
type (
@@ -172,7 +181,9 @@ func (em *entityManager) execute(ctx context.Context, method string, entityPath
172181
}
173182

174183
q := req.Raw().URL.Query()
175-
q.Add("api-version", "2021-05")
184+
// Set rather than Add, so an entityPath that already carries an api-version - an ATOM
185+
// feed href, for example - yields one value rather than two.
186+
q.Set(apiVersionQueryParam, defaultAPIVersion)
176187
req.Raw().URL.RawQuery = q.Encode()
177188

178189
if body != nil {
@@ -244,6 +255,12 @@ func deserializeBody(resp *http.Response, respObj any) (*http.Response, error) {
244255

245256
func newEntityManagerImpl(provider *sbauth.TokenProvider, version string, options *policy.ClientOptions, ns string) (EntityManager, error) {
246257
popts := runtime.PipelineOptions{
258+
// Declaring where the version lives lets azcore replace the default set in
259+
// execute() with policy.ClientOptions.APIVersion, when the caller sets one.
260+
APIVersion: runtime.APIVersionOptions{
261+
Location: runtime.APIVersionLocationQueryParam,
262+
Name: apiVersionQueryParam,
263+
},
247264
PerRetry: []policy.Policy{
248265
&perRetryAuthPolicy{tp: provider},
249266
},

0 commit comments

Comments
 (0)