You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by he...@apache.org on 2024/05/01 22:26:26 UTC
(pulsar-client-go) 02/02: test
This is an automated email from the ASF dual-hosted git repository.
heesung pushed a commit to branch blue-green
in repository https://gitbox.apache.org/repos/asf/pulsar-client-go.git
commit cfd27420e9591361ae2fa99d329d3698f2f98977
Author: Heesung Sohn <he...@streamnative.io>
AuthorDate: Tue Apr 30 18:43:33 2024 -0700
test
---
pulsar/client_impl.go | 21 +-----
pulsar/consumer_partition.go | 9 ++-
pulsar/internal/connection.go | 54 +++++++++++++++
pulsar/internal/lookup_service.go | 31 +++++++--
pulsar/internal/lookup_service_test.go | 18 +++++
pulsar/internal/rpc_client.go | 122 +++++++++++++++++++++++++++++----
pulsar/producer_partition.go | 11 ++-
7 files changed, 224 insertions(+), 42 deletions(-)
diff --git a/pulsar/client_impl.go b/pulsar/client_impl.go
index 8471e1e3..e391c808 100644
--- a/pulsar/client_impl.go
+++ b/pulsar/client_impl.go
@@ -169,26 +169,11 @@ func newClient(options ClientOptions) (Client, error) {
operationTimeout: operationTimeout,
tlsEnabled: tlsConfig != nil,
}
- serviceNameResolver := internal.NewPulsarServiceNameResolver(url)
- c.rpcClient = internal.NewRPCClient(url, serviceNameResolver, c.cnxPool, operationTimeout, logger, metrics)
+ c.rpcClient = internal.NewRPCClient(url, c.cnxPool, operationTimeout, logger, metrics,
+ options.ListenerName, tlsConfig, authProvider)
- switch url.Scheme {
- case "pulsar", "pulsar+ssl":
- c.lookupService = internal.NewLookupService(c.rpcClient, url, serviceNameResolver,
- c.tlsEnabled, options.ListenerName, logger, metrics)
- case "http", "https":
- httpClient, err := internal.NewHTTPClient(url, serviceNameResolver, tlsConfig,
- operationTimeout, logger, metrics, authProvider)
- if err != nil {
- return nil, newError(InvalidConfiguration, fmt.Sprintf("Failed to init http client with err: '%s'",
- err.Error()))
- }
- c.lookupService = internal.NewHTTPLookupService(httpClient, url, serviceNameResolver,
- c.tlsEnabled, logger, metrics)
- default:
- return nil, newError(InvalidConfiguration, fmt.Sprintf("Invalid URL scheme '%s'", url.Scheme))
- }
+ c.lookupService = c.rpcClient.LookupService("")
c.handlers = internal.NewClientHandlers()
diff --git a/pulsar/consumer_partition.go b/pulsar/consumer_partition.go
index dc01e692..977f7ca6 100644
--- a/pulsar/consumer_partition.go
+++ b/pulsar/consumer_partition.go
@@ -176,6 +176,7 @@ type partitionConsumer struct {
ackGroupingTracker ackGroupingTracker
lastMessageInBroker *trackingMessageID
+ redirectedClusterURI string
}
func (pc *partitionConsumer) ActiveConsumerChanged(isActive bool) {
@@ -1371,6 +1372,12 @@ func (pc *partitionConsumer) ConnectionClosed(closeConsumer *pb.CommandCloseCons
}
}
+func (p *partitionConsumer) SetRedirectedClusterURI(redirectedClusterURI string) {
+ // Trigger reconnection in the produce goroutine
+ p.log.WithField("cnx", p._getConn().ID()).Warnf("Set redirectedClusterURI to %s", redirectedClusterURI)
+ p.redirectedClusterURI = redirectedClusterURI
+}
+
// Flow command gives additional permits to send messages to the consumer.
// A typical consumer implementation will use a queue to accumulate these messages
// before the application is ready to consume them. After the consumer is ready,
@@ -1731,7 +1738,7 @@ func (pc *partitionConsumer) reconnectToBroker(connectionClosed *connectionClose
func (pc *partitionConsumer) lookupTopic(brokerServiceURL string) (*internal.LookupResult, error) {
if len(brokerServiceURL) == 0 {
- lr, err := pc.client.lookupService.Lookup(pc.topic)
+ lr, err := pc.client.rpcClient.LookupService(pc.redirectedClusterURI).Lookup(pc.topic)
if err != nil {
pc.log.WithError(err).Warn("Failed to lookup topic")
return nil, err
diff --git a/pulsar/internal/connection.go b/pulsar/internal/connection.go
index 9244328b..65d205b8 100644
--- a/pulsar/internal/connection.go
+++ b/pulsar/internal/connection.go
@@ -70,6 +70,9 @@ type ConnectionListener interface {
// ConnectionClosed close the TCP connection.
ConnectionClosed(closeProducer *pb.CommandCloseProducer)
+
+ // set uri.
+ SetRedirectedClusterURI(redirectedClusterURI string)
}
// Connection is a interface of client cnx.
@@ -94,6 +97,9 @@ type ConsumerHandler interface {
// ConnectionClosed close the TCP connection.
ConnectionClosed(closeConsumer *pb.CommandCloseConsumer)
+
+ // set uri.
+ SetRedirectedClusterURI(redirectedClusterURI string)
}
type connectionState int32
@@ -584,6 +590,9 @@ func (c *connection) internalReceivedCommand(cmd *pb.BaseCommand, headersAndPayl
case pb.BaseCommand_CLOSE_CONSUMER:
c.handleCloseConsumer(cmd.GetCloseConsumer())
+ case pb.BaseCommand_TOPIC_MIGRATED:
+ c.handleTopicMigrated(cmd.GetTopicMigrated())
+
case pb.BaseCommand_AUTH_CHALLENGE:
c.handleAuthChallenge(cmd.GetAuthChallenge())
@@ -928,6 +937,51 @@ func (c *connection) handleCloseProducer(closeProducer *pb.CommandCloseProducer)
}
}
+func (c *connection) getMigratedBrokerServiceUrl(commandTopicMigrated *pb.CommandTopicMigrated) string {
+ if c.tlsOptions == nil {
+ if commandTopicMigrated.GetBrokerServiceUrl() != "" {
+ return commandTopicMigrated.GetBrokerServiceUrl()
+ }
+ } else if commandTopicMigrated.GetBrokerServiceUrlTls() != "" {
+ return commandTopicMigrated.GetBrokerServiceUrlTls()
+ }
+ return ""
+}
+
+func (c *connection) handleTopicMigrated(commandTopicMigrated *pb.CommandTopicMigrated) {
+ resourceID := commandTopicMigrated.GetResourceId()
+ migratedBrokerServiceUrl := c.getMigratedBrokerServiceUrl(commandTopicMigrated)
+ if migratedBrokerServiceUrl == "" {
+ c.log.Warnf("Failed to find the migrated broker url for resource: %s, migratedBrokerUrl: %s, migratedBrokerUrlTls:%s",
+ resourceID,
+ commandTopicMigrated.GetBrokerServiceUrl(),
+ commandTopicMigrated.GetBrokerServiceUrlTls())
+ return
+ }
+ if commandTopicMigrated.ResourceType.Number() == 0 {
+ c.listenersLock.RLock()
+ producer, ok := c.listeners[resourceID]
+ c.listenersLock.RUnlock()
+ if ok {
+ producer.SetRedirectedClusterURI(migratedBrokerServiceUrl)
+ } else {
+ c.log.
+ WithField("producerID", resourceID).
+ Warn("Failed to SetRedirectedClusterURI")
+ }
+ } else {
+ consumer, ok := c.consumerHandler(resourceID)
+ if ok {
+ consumer.SetRedirectedClusterURI(migratedBrokerServiceUrl)
+ } else {
+ c.log.
+ WithField("consumerID", resourceID).
+ Warn("Failed to SetRedirectedClusterURI")
+ }
+ }
+
+}
+
func (c *connection) RegisterListener(id uint64, listener ConnectionListener) error {
// do not add if connection is closed
if c.closed() {
diff --git a/pulsar/internal/lookup_service.go b/pulsar/internal/lookup_service.go
index 8b7fb01b..e71e5003 100644
--- a/pulsar/internal/lookup_service.go
+++ b/pulsar/internal/lookup_service.go
@@ -66,6 +66,8 @@ type LookupService interface {
GetBrokerAddress(brokerServiceURL string, proxyThroughServiceURL bool) (*LookupResult, error)
+ ResolveHost() (*url.URL, error)
+
// Closable Allow Lookup Service's internal client to be able to closed
Closable
}
@@ -137,12 +139,18 @@ const lookupResultMaxRedirect = 20
func (ls *lookupService) Lookup(topic string) (*LookupResult, error) {
ls.metrics.LookupRequestsCount.Inc()
id := ls.rpcClient.NewRequestID()
- res, err := ls.rpcClient.RequestToAnyBroker(id, pb.BaseCommand_LOOKUP, &pb.CommandLookupTopic{
- RequestId: &id,
- Topic: &topic,
- Authoritative: proto.Bool(false),
- AdvertisedListenerName: proto.String(ls.listenerName),
- })
+ host, err := ls.serviceNameResolver.ResolveHost()
+ if err != nil {
+ ls.log.WithError(err).Errorf("rpc client failed to resolve host")
+ return nil, err
+ }
+ res, err := ls.rpcClient.RequestToHost(host, id, pb.BaseCommand_LOOKUP,
+ &pb.CommandLookupTopic{
+ RequestId: &id,
+ Topic: &topic,
+ Authoritative: proto.Bool(false),
+ AdvertisedListenerName: proto.String(ls.listenerName),
+ })
if err != nil {
return nil, err
}
@@ -255,6 +263,11 @@ func (ls *lookupService) GetTopicsOfNamespace(namespace string, mode GetTopicsOf
func (ls *lookupService) Close() {}
+func (ls *lookupService) ResolveHost() (*url.URL, error) {
+ return ls.serviceNameResolver.ResolveHost()
+}
+
+
const HTTPLookupServiceBasePathV1 string = "/lookup/v2/destination/"
const HTTPLookupServiceBasePathV2 string = "/lookup/v2/topic/"
const HTTPAdminServiceV1Format string = "/admin/%s/partitions"
@@ -362,6 +375,11 @@ func (h *httpLookupService) GetTopicsOfNamespace(namespace string, mode GetTopic
func (h *httpLookupService) GetSchema(topic string, schemaVersion []byte) (schema *pb.Schema, err error) {
return nil, errors.New("GetSchema is not supported by httpLookupService")
}
+
+func (h *httpLookupService) ResolveHost() (*url.URL, error) {
+ return h.serviceNameResolver.ResolveHost()
+}
+
func (h *httpLookupService) Close() {
h.httpClient.Close()
}
@@ -385,3 +403,4 @@ func selectServiceURL(tlsEnabled bool, brokerServiceURL, brokerServiceURLTLS str
}
return brokerServiceURL
}
+
diff --git a/pulsar/internal/lookup_service_test.go b/pulsar/internal/lookup_service_test.go
index 1504f3f2..d38bef2d 100644
--- a/pulsar/internal/lookup_service_test.go
+++ b/pulsar/internal/lookup_service_test.go
@@ -77,6 +77,15 @@ func (c *mockedLookupRPCClient) RequestToAnyBroker(requestID uint64, cmdType pb.
}, nil
}
+func (c *mockedLookupRPCClient) RequestToHost(host *url.URL, requestID uint64,
+ cmdType pb.BaseCommand_Type, message proto.Message) (*RPCResult, error) {
+ return c.RequestToAnyBroker(requestID, cmdType, message)
+}
+
+func (c *mockedLookupRPCClient) LookupService(URL string) LookupService {
+ return nil
+}
+
func (c *mockedLookupRPCClient) Request(logicalAddr *url.URL, physicalAddr *url.URL, requestID uint64,
cmdType pb.BaseCommand_Type, message proto.Message) (*RPCResult, error) {
assert.Equal(c.t, cmdType, pb.BaseCommand_LOOKUP)
@@ -486,6 +495,15 @@ func (m mockedPartitionedTopicMetadataRPCClient) RequestOnCnx(cnx Connection, re
return nil, nil
}
+func (c *mockedPartitionedTopicMetadataRPCClient) RequestToHost(host *url.URL, requestID uint64,
+ cmdType pb.BaseCommand_Type, message proto.Message) (*RPCResult, error) {
+ return c.RequestToAnyBroker(requestID, cmdType, message)
+}
+
+func (c *mockedPartitionedTopicMetadataRPCClient) LookupService(URL string) LookupService {
+ return nil
+}
+
func TestGetPartitionedTopicMetadataSuccess(t *testing.T) {
url, err := url.Parse("pulsar://example:6650")
assert.NoError(t, err)
diff --git a/pulsar/internal/rpc_client.go b/pulsar/internal/rpc_client.go
index 2213083d..35dcab06 100644
--- a/pulsar/internal/rpc_client.go
+++ b/pulsar/internal/rpc_client.go
@@ -19,7 +19,10 @@ package internal
import (
"errors"
+ "fmt"
+ "github.com/apache/pulsar-client-go/pulsar/auth"
"net/url"
+ "sync"
"sync/atomic"
"time"
@@ -55,12 +58,16 @@ type RPCClient interface {
// Send a request and block until the result is available
RequestToAnyBroker(requestID uint64, cmdType pb.BaseCommand_Type, message proto.Message) (*RPCResult, error)
+ RequestToHost(host *url.URL, requestID uint64, cmdType pb.BaseCommand_Type, message proto.Message) (*RPCResult, error)
+
Request(logicalAddr *url.URL, physicalAddr *url.URL, requestID uint64,
cmdType pb.BaseCommand_Type, message proto.Message) (*RPCResult, error)
RequestOnCnxNoWait(cnx Connection, cmdType pb.BaseCommand_Type, message proto.Message) error
RequestOnCnx(cnx Connection, requestID uint64, cmdType pb.BaseCommand_Type, message proto.Message) (*RPCResult, error)
+
+ LookupService(URL string) LookupService
}
type rpcClient struct {
@@ -72,34 +79,43 @@ type rpcClient struct {
consumerIDGenerator uint64
log log.Logger
metrics *Metrics
+ tlsConfig *TLSOptions
+ listenerName string
+ authProvider auth.Provider
+ lookupService LookupService
+ urlLookupServiceMapLock sync.RWMutex
+ urlLookupServiceMap map[string]LookupService
}
-func NewRPCClient(serviceURL *url.URL, serviceNameResolver ServiceNameResolver, pool ConnectionPool,
- requestTimeout time.Duration, logger log.Logger, metrics *Metrics) RPCClient {
- return &rpcClient{
- serviceNameResolver: serviceNameResolver,
- pool: pool,
- requestTimeout: requestTimeout,
- log: logger.SubLogger(log.Fields{"serviceURL": serviceURL}),
- metrics: metrics,
+func NewRPCClient(serviceURL *url.URL, pool ConnectionPool,
+ requestTimeout time.Duration, logger log.Logger, metrics *Metrics, listenerName string, tlsConfig *TLSOptions, authProvider auth.Provider) RPCClient {
+ c := rpcClient{
+ pool: pool,
+ requestTimeout: requestTimeout,
+ log: logger.SubLogger(log.Fields{"serviceURL": serviceURL}),
+ metrics: metrics,
+ listenerName: listenerName,
+ tlsConfig: tlsConfig,
+ authProvider: authProvider,
+ }
+ lookupService, err := c.NewLookupService(serviceURL)
+ if err != nil {
+ panic(err)
}
+ c.lookupService = lookupService
+
+ return &c
}
-func (c *rpcClient) RequestToAnyBroker(requestID uint64, cmdType pb.BaseCommand_Type,
+func (c *rpcClient) requestToHost(host *url.URL, requestID uint64, cmdType pb.BaseCommand_Type,
message proto.Message) (*RPCResult, error) {
var err error
- var host *url.URL
var rpcResult *RPCResult
startTime := time.Now()
backoff := DefaultBackoff{100 * time.Millisecond}
// we can retry these requests because this kind of request is
// not specific to any particular broker
for time.Since(startTime) < c.requestTimeout {
- host, err = c.serviceNameResolver.ResolveHost()
- if err != nil {
- c.log.WithError(err).Errorf("rpc client failed to resolve host")
- return nil, err
- }
rpcResult, err = c.Request(host, host, requestID, cmdType, message)
// success we got a response
if err == nil {
@@ -114,6 +130,26 @@ func (c *rpcClient) RequestToAnyBroker(requestID uint64, cmdType pb.BaseCommand_
return rpcResult, err
}
+func (c *rpcClient) RequestToAnyBroker(requestID uint64, cmdType pb.BaseCommand_Type,
+ message proto.Message) (*RPCResult, error) {
+ var err error
+ var host *url.URL
+
+ host, err = c.serviceNameResolver.ResolveHost()
+ if err != nil {
+ c.log.WithError(err).Errorf("rpc client failed to resolve host")
+ return nil, err
+ }
+
+ return c.requestToHost(host, requestID, cmdType, message)
+}
+
+func (c *rpcClient) RequestToHost(host *url.URL, requestID uint64,
+ cmdType pb.BaseCommand_Type, message proto.Message) (*RPCResult, error) {
+ return c.requestToHost(host, requestID, cmdType, message)
+}
+
+
func (c *rpcClient) Request(logicalAddr *url.URL, physicalAddr *url.URL, requestID uint64,
cmdType pb.BaseCommand_Type, message proto.Message) (*RPCResult, error) {
c.metrics.RPCRequestCount.Inc()
@@ -188,3 +224,59 @@ func (c *rpcClient) NewProducerID() uint64 {
func (c *rpcClient) NewConsumerID() uint64 {
return atomic.AddUint64(&c.consumerIDGenerator, 1)
}
+
+func (c *rpcClient) LookupService(URL string) LookupService {
+ if URL == "" {
+ return c.lookupService
+ }
+ c.urlLookupServiceMapLock.Lock()
+ defer c.urlLookupServiceMapLock.Unlock()
+ lookupService, ok := c.urlLookupServiceMap[URL]
+ if ok {
+ return lookupService
+ }
+
+ serviceUrl, err := url.Parse(URL)
+ if err != nil {
+ panic(err)
+ }
+
+ lookupService, err2 := c.NewLookupService(serviceUrl)
+ if err != nil {
+ panic(err2)
+ }
+ c.urlLookupServiceMap[URL] = lookupService
+ return lookupService
+
+}
+
+func (c *rpcClient) NewLookupService(url *url.URL) (LookupService, error) {
+
+ switch url.Scheme {
+ case "pulsar", "pulsar+ssl":
+ c.serviceNameResolver = NewPulsarServiceNameResolver(url)
+ return NewLookupService(c, url, c.serviceNameResolver,
+ c.tlsConfig != nil, c.listenerName, c.log, c.metrics), nil
+ case "http", "https":
+ c.serviceNameResolver = NewPulsarServiceNameResolver(url)
+ httpClient, err := NewHTTPClient(url, c.serviceNameResolver, c.tlsConfig,
+ c.requestTimeout, c.log, c.metrics, c.authProvider)
+ if err != nil {
+ return nil, err
+ }
+
+ return NewHTTPLookupService(
+ httpClient, url, c.serviceNameResolver, c.tlsConfig != nil, c.log, c.metrics), nil
+ default:
+ panic(fmt.Sprintf("Invalid URL scheme '%s'", url.Scheme))
+ }
+}
+
+func (c *rpcClient) Close() {
+ c.lookupService.Close()
+ c.urlLookupServiceMapLock.Lock()
+ defer c.urlLookupServiceMapLock.Unlock()
+ for _, value := range c.urlLookupServiceMap {
+ value.Close()
+ }
+}
diff --git a/pulsar/producer_partition.go b/pulsar/producer_partition.go
index f5c838c4..8dbb3cc6 100755
--- a/pulsar/producer_partition.go
+++ b/pulsar/producer_partition.go
@@ -118,6 +118,7 @@ type partitionProducer struct {
epoch uint64
schemaCache *schemaCache
topicEpoch *uint64
+ redirectedClusterURI string
}
type schemaCache struct {
@@ -223,7 +224,7 @@ func newPartitionProducer(client *client, topic string, options *ProducerOptions
func (p *partitionProducer) lookupTopic(brokerServiceURL string) (*internal.LookupResult, error) {
if len(brokerServiceURL) == 0 {
- lr, err := p.client.lookupService.Lookup(p.topic)
+ lr, err := p.client.rpcClient.LookupService(p.redirectedClusterURI).Lookup(p.topic)
if err != nil {
p.log.WithError(err).Warn("Failed to lookup topic")
return nil, err
@@ -232,7 +233,7 @@ func (p *partitionProducer) lookupTopic(brokerServiceURL string) (*internal.Look
p.log.Debug("Lookup result: ", lr)
return lr, err
}
- return p.client.lookupService.GetBrokerAddress(brokerServiceURL, p._getConn().IsProxied())
+ return p.client.rpcClient.LookupService(p.redirectedClusterURI).GetBrokerAddress(brokerServiceURL, p._getConn().IsProxied())
}
func (p *partitionProducer) grabCnx(assignedBrokerURL string) error {
@@ -403,6 +404,12 @@ func (p *partitionProducer) ConnectionClosed(closeProducer *pb.CommandCloseProdu
}
}
+func (p *partitionProducer) SetRedirectedClusterURI(redirectedClusterURI string) {
+ // Trigger reconnection in the produce goroutine
+ p.log.WithField("cnx", p._getConn().ID()).Warnf("Set redirectedClusterURI to %s", redirectedClusterURI)
+ p.redirectedClusterURI = redirectedClusterURI
+}
+
func (p *partitionProducer) getOrCreateSchema(schemaInfo *SchemaInfo) (schemaVersion []byte, err error) {
tmpSchemaType := pb.Schema_Type(int32(schemaInfo.Type))