fix: replace panic with proper error handling in RPC client and lookup service

Author: RobertIndie

pulsar/client_impl.go

@@ -171,10 +171,16 @@ func newClient(options ClientOptions) (Client, error) {
 		tlsEnabled:       tlsConfig != nil,
 	}
 
-	c.rpcClient = internal.NewRPCClient(url, c.cnxPool, operationTimeout, logger, metrics,
+	c.rpcClient, err = internal.NewRPCClient(url, c.cnxPool, operationTimeout, logger, metrics,
 		options.ListenerName, tlsConfig, authProvider, toKeyValues(options.LookupProperties))
+	if err != nil {
+		return nil, err
+	}
 
-	c.lookupService = c.rpcClient.LookupService("")
+	c.lookupService, err = c.rpcClient.LookupService("")
+	if err != nil {
+		return nil, err
+	}
 
 	c.handlers = internal.NewClientHandlers()
 

pulsar/consumer_partition.go

@@ -1934,7 +1934,11 @@ func (pc *partitionConsumer) reconnectToBroker(connectionClosed *connectionClose
 
 func (pc *partitionConsumer) lookupTopic(brokerServiceURL string) (*internal.LookupResult, error) {
 	if len(brokerServiceURL) == 0 {
-		lr, err := pc.client.rpcClient.LookupService(pc.redirectedClusterURI).Lookup(pc.topic)
+		lookupService, err := pc.client.rpcClient.LookupService(pc.redirectedClusterURI)
+		if err != nil {
+			return nil, fmt.Errorf("failed to get lookup service: %w", err)
+		}
+		lr, err := lookupService.Lookup(pc.topic)
 		if err != nil {
 			pc.log.WithError(err).Warn("Failed to lookup topic")
 			return nil, err

pulsar/internal/lookup_service_test.go

@@ -82,8 +82,8 @@ func (c *mockedLookupRPCClient) RequestToHost(_ *ServiceNameResolver, requestID
 	return c.RequestToAnyBroker(requestID, cmdType, message)
 }
 
-func (c *mockedLookupRPCClient) LookupService(_ string) LookupService {
-	return nil
+func (c *mockedLookupRPCClient) LookupService(_ string) (LookupService, error) {
+	return nil, nil
 }
 
 func (c *mockedLookupRPCClient) Request(logicalAddr *url.URL, physicalAddr *url.URL, _ uint64,
@@ -521,8 +521,8 @@ func (m *mockedPartitionedTopicMetadataRPCClient) RequestToHost(_ *ServiceNameRe
 	return m.RequestToAnyBroker(requestID, cmdType, message)
 }
 
-func (m *mockedPartitionedTopicMetadataRPCClient) LookupService(_ string) LookupService {
-	return nil
+func (m *mockedPartitionedTopicMetadataRPCClient) LookupService(_ string) (LookupService, error) {
+	return nil, nil
 }
 
 func TestGetPartitionedTopicMetadataSuccess(t *testing.T) {

pulsar/internal/rpc_client.go

@@ -74,7 +74,7 @@ type RPCClient interface {
 
 	RequestOnCnx(cnx Connection, requestID uint64, cmdType pb.BaseCommand_Type, message proto.Message) (*RPCResult, error)
 
-	LookupService(URL string) LookupService
+	LookupService(URL string) (LookupService, error)
 }
 
 type rpcClient struct {
@@ -96,7 +96,7 @@ type rpcClient struct {
 
 func NewRPCClient(serviceURL *url.URL, pool ConnectionPool,
 	requestTimeout time.Duration, logger log.Logger, metrics *Metrics,
-	listenerName string, tlsConfig *TLSOptions, authProvider auth.Provider, lookupProperties []*pb.KeyValue) RPCClient {
+	listenerName string, tlsConfig *TLSOptions, authProvider auth.Provider, lookupProperties []*pb.KeyValue) (RPCClient, error) {
 	c := rpcClient{
 		pool:                pool,
 		requestTimeout:      requestTimeout,
@@ -110,11 +110,11 @@ func NewRPCClient(serviceURL *url.URL, pool ConnectionPool,
 	}
 	lookupService, err := c.NewLookupService(serviceURL)
 	if err != nil {
-		panic(err)
+		return nil, fmt.Errorf("failed to create lookup service: %w", err)
 	}
 	c.lookupService = lookupService
 
-	return &c
+	return &c, nil
 }
 
 func (c *rpcClient) requestToHost(serviceNameResolver *ServiceNameResolver,
@@ -220,29 +220,28 @@ func (c *rpcClient) NewConsumerID() uint64 {
 	return atomic.AddUint64(&c.consumerIDGenerator, 1)
 }
 
-func (c *rpcClient) LookupService(URL string) LookupService {
+func (c *rpcClient) LookupService(URL string) (LookupService, error) {
 	if URL == "" {
-		return c.lookupService
+		return c.lookupService, nil
 	}
 	c.urlLookupServiceMapLock.Lock()
 	defer c.urlLookupServiceMapLock.Unlock()
 	lookupService, ok := c.urlLookupServiceMap[URL]
 	if ok {
-		return lookupService
+		return lookupService, nil
 	}
 
 	serviceURL, err := url.Parse(URL)
 	if err != nil {
-		panic(err)
+		return nil, fmt.Errorf("failed to parse URL '%s': %w", URL, err)
 	}
 
 	lookupService, err = c.NewLookupService(serviceURL)
 	if err != nil {
-		panic(err)
+		return nil, fmt.Errorf("failed to create lookup service for URL '%s': %w", URL, err)
 	}
 	c.urlLookupServiceMap[URL] = lookupService
-	return lookupService
-
+	return lookupService, nil
 }
 
 func (c *rpcClient) NewLookupService(url *url.URL) (LookupService, error) {
@@ -263,7 +262,7 @@ func (c *rpcClient) NewLookupService(url *url.URL) (LookupService, error) {
 		return NewHTTPLookupService(
 			httpClient, url, serviceNameResolver, c.tlsConfig != nil, c.log, c.metrics), nil
 	default:
-		panic(fmt.Sprintf("Invalid URL scheme '%s'", url.Scheme))
+		return nil, fmt.Errorf("Invalid URL scheme '%s'", url.Scheme)
 	}
 }
 

pulsar/internal/rpc_client_test.go

@@ -0,0 +1,59 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//	http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+package internal
+
+import (
+	"net/url"
+	"testing"
+
+	"github.com/apache/pulsar-client-go/pulsar/log"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestNewRPCClient_InvalidURL_ShouldNotPanic(t *testing.T) {
+	// Test that NewRPCClient doesn't panic with invalid URL
+	invalidURL, _ := url.Parse("invalid://scheme")
+
+	// This should not panic and should return an error
+	_, err := NewRPCClient(invalidURL, nil, 0, log.DefaultNopLogger(), nil, "", nil, nil, nil)
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "Invalid URL scheme")
+}
+
+func TestLookupService_InvalidURL_ShouldNotPanic(t *testing.T) {
+	// Create a minimal RPC client for testing
+	validURL, _ := url.Parse("pulsar://localhost:6650")
+	rpcClient, err := NewRPCClient(validURL, nil, 0, log.DefaultNopLogger(), nil, "", nil, nil, nil)
+	assert.NoError(t, err)
+
+	// Test that LookupService doesn't panic with invalid URL
+	_, err = rpcClient.LookupService("invalid://url")
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "Invalid URL scheme")
+}
+
+func TestLookupService_InvalidScheme_ShouldNotPanic(t *testing.T) {
+	// Create a minimal RPC client for testing
+	validURL, _ := url.Parse("pulsar://localhost:6650")
+	rpcClient, err := NewRPCClient(validURL, nil, 0, log.DefaultNopLogger(), nil, "", nil, nil, nil)
+	assert.NoError(t, err)
+
+	// Test that LookupService doesn't panic with invalid scheme
+	_, err = rpcClient.LookupService("ftp://localhost:21")
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "Invalid URL scheme")
+}

pulsar/producer_partition.go

@@ -244,7 +244,11 @@ 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.rpcClient.LookupService(p.redirectedClusterURI).Lookup(p.topic)
+		lookupService, err := p.client.rpcClient.LookupService(p.redirectedClusterURI)
+		if err != nil {
+			return nil, fmt.Errorf("failed to get lookup service: %w", err)
+		}
+		lr, err := lookupService.Lookup(p.topic)
 		if err != nil {
 			p.log.WithError(err).Warn("Failed to lookup topic")
 			return nil, err
@@ -253,8 +257,11 @@ func (p *partitionProducer) lookupTopic(brokerServiceURL string) (*internal.Look
 		p.log.Debug("Lookup result: ", lr)
 		return lr, err
 	}
-	return p.client.rpcClient.LookupService(p.redirectedClusterURI).
-		GetBrokerAddress(brokerServiceURL, p._getConn().IsProxied())
+	lookupService, err := p.client.rpcClient.LookupService(p.redirectedClusterURI)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get lookup service: %w", err)
+	}
+	return lookupService.GetBrokerAddress(brokerServiceURL, p._getConn().IsProxied())
 }
 
 func (p *partitionProducer) grabCnx(assignedBrokerURL string) error {