fix: replace panic with proper error handling in RPC client and lookup service (#1403)

### Motivation

Fixes #1402

The current implementation uses `panic()` calls in several critical code paths within the RPC client and lookup service, which can cause the entire application to crash when encountering invalid URLs or configuration errors. This is not a good practice for a client library as it makes the application unstable and difficult to handle gracefully.

The main issues addressed:
- `NewRPCClient()` panics when creating lookup service fails
- `LookupService()` panics when parsing invalid URLs or creating lookup services
- `NewLookupService()` panics for invalid URL schemes

### Modifications

- **RPC Client Interface**: Modified `LookupService()` method signature to return `(LookupService, error)` instead of just `LookupService`
- **NewRPCClient**: Changed to return `(RPCClient, error)` and handle lookup service creation errors properly
- **LookupService Implementation**: Replaced all `panic()` calls with proper error returns and descriptive error

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,8 @@ 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 +111,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 +221,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 +263,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,60 @@
+// 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 {