You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by zi...@apache.org on 2026/10/20 04:04:13 UTC
(pulsar-client-go) branch master updated: [fix][test] Make admin and transaction tests tolerant of broker behavior changes (#1545)
This is an automated email from the ASF dual-hosted git repository.
nodece pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar-client-go.git
The following commit(s) were added to refs/heads/master by this push:
new e3619f39 [fix][test] Make admin and transaction tests tolerant of broker behavior changes (#1545)
e3619f39 is described below
commit e3619f396655d8e26d66f146fd858cca5083d2d1
Author: crossoverJie <cr...@gmail.com>
AuthorDate: Tue Oct 20 12:04:08 2026 +0800
[fix][test] Make admin and transaction tests tolerant of broker behavior changes (#1545)
* [fix][test] Make TestSetTopicAutoCreation tolerant of broker error message changes
The test asserted exact broker error strings, which broke when Pulsar
changed the messages:
- apache/pulsar#25163 (shipped in 4.2.5) relaxed the non-partitioned
autoTopicCreation override validation and changed the rejection text from
"[defaultNumPartitions] is not allowed to be set when the type is
non-partition." to "[defaultNumPartitions] must be null, 0 or 1 when the
type is non-partitioned."
- 5.0.0 changed the missing-tenant error from "Tenant does not exist" to
"Namespace does not exist".
Replace the single exact-match `errReason` with an `errReasons` slice and
assert containment, so the test passes against both old and new broker
versions instead of being brittle to broker wording changes.
Verified against Pulsar 4.2.4, 4.2.5 and 5.0.0.
Co-Authored-By: CodeBuddy Code <no...@codebuddy.ai>
* [fix][test] Authenticate transaction metadata request as super-user
TestTxn_TCClient read transaction metadata through the raw admin HTTP helper
(httpGet -> httpDo), which calls the plain-text web service without any
credentials, so the broker authenticates it as the "anonymous" role.
apache/pulsar#26773 ("Improve transaction admin checks", shipped in 4.2.5 and
5.0.0) added validateSuperUserAccess() to the transaction metadata admin
endpoint (GET /admin/v3/transactions/transactionMetadata/{mostSigBits}/{leastSigBits}).
The unauthenticated request is now rejected with HTTP 401, so transactionStats
fails and the test panics on the nil stats map.
Add an httpGetWithTLS helper that calls the TLS web service using the test
client certificate, whose CN ("admin") is a super-user role in standalone.conf,
and use it for the transaction metadata request.
Verified against Pulsar 4.2.4, 4.2.5 and 5.0.0.
Co-Authored-By: CodeBuddy Code <no...@codebuddy.ai>
* [fix][test] Make TestPartitionState tolerant of removed stats field
apache/pulsar#26687 (PIP-379: Remove the classic Shared and Key_Shared
dispatcher implementations, shipped in 5.0.0) removed
consumersAfterMarkDeletePosition from the subscription stats. The broker no
longer serializes the field, so it unmarshals to a nil map instead of the
empty map the test asserted.
Assert the map is empty, which holds both for the old behavior (empty map)
and the new one (absent/nil).
Verified against Pulsar 4.2.4, 4.2.5 and 5.0.0.
Co-Authored-By: CodeBuddy Code <no...@codebuddy.ai>
* [fix][test] Address review feedback on admin test helpers
- httpGetWithTLS: give the HTTPS test client a 30s timeout so a stalled
broker/network path cannot hang the test run.
- TestSetTopicAutoCreation: replace the unchecked `err.(rest.Error)` type
assertion (which panics on transport/TLS errors) with `require.ErrorAs`.
Co-Authored-By: CodeBuddy Code <no...@codebuddy.ai>
* [fix][test] Cache the TLS admin HTTP client in test helper
Build the HTTPS client (key pair, CA pool) once via sync.Once and reuse it
across calls instead of rebuilding it on every request.
Co-Authored-By: CodeBuddy Code <no...@codebuddy.ai>
---------
Co-authored-by: CodeBuddy Code <no...@codebuddy.ai>
---
pulsar/helper_for_test.go | 80 ++++++++++++++++++++++++++++++++-
pulsaradmin/pkg/admin/namespace_test.go | 49 +++++++++++---------
pulsaradmin/pkg/admin/topic_test.go | 4 +-
3 files changed, 109 insertions(+), 24 deletions(-)
diff --git a/pulsar/helper_for_test.go b/pulsar/helper_for_test.go
index 1cb3f6cb..065b04f9 100644
--- a/pulsar/helper_for_test.go
+++ b/pulsar/helper_for_test.go
@@ -19,12 +19,16 @@ package pulsar
import (
"bytes"
+ "crypto/tls"
+ "crypto/x509"
"encoding/json"
"fmt"
"io"
"net/http"
+ "os"
"path"
"strings"
+ "sync"
"testing"
"time"
@@ -157,13 +161,87 @@ func topicStats(topic string) (map[string]interface{}, error) {
return metadata, err
}
+// transactionStats queries the transaction metadata admin endpoint. Since
+// apache/pulsar#26773 that endpoint requires super-user access, so the request must be
+// authenticated. The plain-text httpGet helper sends no credentials and would be rejected
+// as the "anonymous" role (HTTP 401), so use the TLS web service with the test client
+// certificate instead, whose CN ("admin") is a super-user role in standalone.conf.
func transactionStats(id *TxnID) (map[string]interface{}, error) {
var metadata map[string]interface{}
path := fmt.Sprintf("admin/v3/transactions/transactionMetadata/%d/%d", id.MostSigBits, id.LeastSigBits)
- err := httpGet(path, &metadata)
+ err := httpGetWithTLS(path, &metadata)
return metadata, err
}
+// httpGetWithTLS performs an HTTPS GET against the broker's web service, authenticating
+// with the test client certificate. It is used for admin endpoints that require a
+// super-user role.
+func httpGetWithTLS(requestPath string, out interface{}) error {
+ client, err := getTLSWebClient()
+ if err != nil {
+ return err
+ }
+ req, err := http.NewRequest(http.MethodGet, webServiceURLTLS+"/"+requestPath, nil)
+ if err != nil {
+ return err
+ }
+ req.Header = jsonHeaders()
+ resp, err := client.Do(req)
+ if err != nil {
+ return err
+ }
+ defer resp.Body.Close()
+ if resp.StatusCode > 299 || resp.StatusCode < 200 {
+ return fmt.Errorf("http error status code: %d", resp.StatusCode)
+ }
+ if out == nil {
+ return nil
+ }
+ outBytes, err := io.ReadAll(resp.Body)
+ if err != nil {
+ return err
+ }
+ return json.Unmarshal(outBytes, out)
+}
+
+var (
+ tlsWebClientOnce sync.Once
+ tlsWebClient *http.Client
+ tlsWebClientErr error
+)
+
+// getTLSWebClient lazily builds and caches an HTTPS client that authenticates with the
+// test client certificate. It is used for admin endpoints that require a super-user role.
+func getTLSWebClient() (*http.Client, error) {
+ tlsWebClientOnce.Do(func() {
+ cert, err := tls.LoadX509KeyPair(tlsClientCertPath, tlsClientKeyPath)
+ if err != nil {
+ tlsWebClientErr = err
+ return
+ }
+ caCert, err := os.ReadFile(caCertsPath)
+ if err != nil {
+ tlsWebClientErr = err
+ return
+ }
+ pool := x509.NewCertPool()
+ if !pool.AppendCertsFromPEM(caCert) {
+ tlsWebClientErr = fmt.Errorf("failed to parse CA certificates from %s", caCertsPath)
+ return
+ }
+ tlsWebClient = &http.Client{
+ Timeout: 30 * time.Second,
+ Transport: &http.Transport{
+ TLSClientConfig: &tls.Config{
+ Certificates: []tls.Certificate{cert},
+ RootCAs: pool,
+ },
+ },
+ }
+ })
+ return tlsWebClient, tlsWebClientErr
+}
+
func topicPath(topic string) string {
tn, _ := internal.ParseTopicName(topic)
idx := strings.LastIndex(tn.Name, "/")
diff --git a/pulsaradmin/pkg/admin/namespace_test.go b/pulsaradmin/pkg/admin/namespace_test.go
index 8f3131b9..a0166b66 100644
--- a/pulsaradmin/pkg/admin/namespace_test.go
+++ b/pulsaradmin/pkg/admin/namespace_test.go
@@ -157,10 +157,10 @@ func TestSetTopicAutoCreation(t *testing.T) {
require.NotNil(t, admin)
tests := []struct {
- name string
- namespace string
- config utils.TopicAutoCreationConfig
- errReason string
+ name string
+ namespace string
+ config utils.TopicAutoCreationConfig
+ errReasons []string // accepted error reasons; empty means success is expected
}{
{
name: "Set partitioned type topic auto creation",
@@ -170,7 +170,7 @@ func TestSetTopicAutoCreation(t *testing.T) {
Type: utils.Partitioned,
Partitions: ptr(3),
},
- errReason: "",
+ errReasons: nil,
},
{
name: "Set partitioned type topic auto creation without partitions",
@@ -179,8 +179,8 @@ func TestSetTopicAutoCreation(t *testing.T) {
Allow: true,
Type: utils.Partitioned,
},
- errReason: "Invalid configuration for autoTopicCreationOverride. the detail is [defaultNumPartitions] " +
- "cannot be null when the type is partitioned.",
+ errReasons: []string{"Invalid configuration for autoTopicCreationOverride. the detail is " +
+ "[defaultNumPartitions] cannot be null when the type is partitioned."},
},
{
name: "Set partitioned type topic auto creation with partitions < 1",
@@ -190,8 +190,8 @@ func TestSetTopicAutoCreation(t *testing.T) {
Type: utils.Partitioned,
Partitions: ptr(-1),
},
- errReason: "Invalid configuration for autoTopicCreationOverride. the detail is [defaultNumPartitions] " +
- "cannot be less than 1 for partition type.",
+ errReasons: []string{"Invalid configuration for autoTopicCreationOverride. the detail is " +
+ "[defaultNumPartitions] cannot be less than 1 for partition type."},
},
{
name: "Set non-partitioned type topic auto creation",
@@ -200,7 +200,7 @@ func TestSetTopicAutoCreation(t *testing.T) {
Allow: true,
Type: utils.NonPartitioned,
},
- errReason: "",
+ errReasons: nil,
},
{
name: "Set non-partitioned type topic auto creation with partitions",
@@ -210,8 +210,12 @@ func TestSetTopicAutoCreation(t *testing.T) {
Type: utils.NonPartitioned,
Partitions: ptr(3),
},
- errReason: "Invalid configuration for autoTopicCreationOverride. the detail is [defaultNumPartitions] is " +
- "not allowed to be set when the type is non-partition.",
+ errReasons: []string{
+ "Invalid configuration for autoTopicCreationOverride. the detail is [defaultNumPartitions] is " +
+ "not allowed to be set when the type is non-partition.",
+ "Invalid configuration for autoTopicCreationOverride. the detail is [defaultNumPartitions] " +
+ "must be null, 0 or 1 when the type is non-partitioned.",
+ },
},
{
name: "Disable topic auto creation",
@@ -219,7 +223,7 @@ func TestSetTopicAutoCreation(t *testing.T) {
config: utils.TopicAutoCreationConfig{
Allow: false,
},
- errReason: "",
+ errReasons: nil,
},
{
name: "Set topic auto creation on a non-exist namespace",
@@ -228,7 +232,7 @@ func TestSetTopicAutoCreation(t *testing.T) {
Allow: true,
Type: utils.NonPartitioned,
},
- errReason: "Namespace does not exist",
+ errReasons: []string{"Namespace does not exist"},
},
{
name: "Set topic auto creation on a non-exist tenant",
@@ -237,23 +241,24 @@ func TestSetTopicAutoCreation(t *testing.T) {
Allow: true,
Type: utils.NonPartitioned,
},
- errReason: "Tenant does not exist",
+ errReasons: []string{"Tenant does not exist", "Namespace does not exist"},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
namespace, _ := utils.GetNamespaceName(tt.namespace)
err := admin.Namespaces().SetTopicAutoCreation(*namespace, tt.config)
- if tt.errReason == "" {
- assert.Equal(t, nil, err)
+ if len(tt.errReasons) == 0 {
+ assert.NoError(t, err)
err = admin.Namespaces().RemoveTopicAutoCreation(*namespace)
- assert.Equal(t, nil, err)
- }
- if err != nil {
- restError := err.(rest.Error)
- assert.Equal(t, tt.errReason, restError.Reason)
+ assert.NoError(t, err)
+ return
}
+ require.Error(t, err)
+ var restError rest.Error
+ require.ErrorAs(t, err, &restError)
+ assert.Contains(t, tt.errReasons, restError.Reason)
})
}
}
diff --git a/pulsaradmin/pkg/admin/topic_test.go b/pulsaradmin/pkg/admin/topic_test.go
index 0a062b4c..6ff88b0d 100644
--- a/pulsaradmin/pkg/admin/topic_test.go
+++ b/pulsaradmin/pkg/admin/topic_test.go
@@ -228,7 +228,9 @@ func TestPartitionState(t *testing.T) {
assert.Equal(t, sub.LastMarkDeleteAdvancedTimestamp, int64(0))
assert.Equal(t, sub.IsDurable, true)
assert.Equal(t, sub.AllowOutOfOrderDelivery, false)
- assert.Equal(t, sub.ConsumersAfterMarkDeletePosition, map[string]string{})
+ // apache/pulsar#26687 (PIP-379, Pulsar 5.0) removed consumersAfterMarkDeletePosition
+ // from the subscription stats, so the field may be absent (nil) instead of an empty map.
+ assert.Empty(t, sub.ConsumersAfterMarkDeletePosition)
assert.Equal(t, sub.NonContiguousDeletedMessagesRanges, 0)
assert.Equal(t, sub.NonContiguousDeletedMessagesRangesSrzSize, 0)
assert.Equal(t, sub.DelayedMessageIndexSizeInBytes, int64(0))