diff --git a/examples.go b/examples.go
index ebd518527..f1d14a9e6 100644
--- a/examples.go
+++ b/examples.go
@@ -19,7 +19,6 @@ package main
 
 import (
 	"fmt"
-	"net/http"
 
 	"github.com/streamnative/pulsarctl/pkg/pulsar"
 )
@@ -28,13 +27,15 @@ import (
 func Examples() {
 	config := &pulsar.Config{
 		WebServiceURL: "http://localhost:8080",
-		HTTPClient:    http.DefaultClient,
 
 		// If the server enable the TLSAuth
-		// Auth: auth.NewAuthenticationTLS()
+		// TLSCertFile: filepath,
+		// TLSKeyFile: key_filepath,
+		// TLSAllowInsecureConnection: true,
 
 		// If the server enable the TokenAuth
-		// TokenAuth: auth.NewAuthenticationToken()
+		// Token: token_string,
+		// TokenFile: toke_filepath,
 	}
 
 	// the default NewPulsarClient will use v2 APIs. If you need to request other version APIs,
diff --git a/pkg/auth/auth_provider.go b/pkg/auth/auth_provider.go
new file mode 100644
index 000000000..f4d595ea8
--- /dev/null
+++ b/pkg/auth/auth_provider.go
@@ -0,0 +1,27 @@
+// 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 auth
+
+// Provider provide a general method to add auth message
+type Provider interface {
+	// HasDataForHTTP is used to check if data for HTTP are available
+	HasDataForHTTP() bool
+
+	// GetHTTPHeaders is used to get all auth headers
+	GetHTTPHeaders() (map[string]string, error)
+}
diff --git a/pkg/auth/tls.go b/pkg/auth/tls.go
index ebe33d683..1f2a406f2 100644
--- a/pkg/auth/tls.go
+++ b/pkg/auth/tls.go
@@ -17,26 +17,28 @@
 
 package auth
 
-import "crypto/tls"
+import (
+	"crypto/tls"
+	"crypto/x509"
+	"io/ioutil"
 
-type TLSAuthProvider struct {
-	certificatePath string
-	privateKeyPath  string
-}
+	"github.com/pkg/errors"
+)
 
-// NewAuthenticationTLSWithParams initialize the authentication provider with map param.
-func NewAuthenticationTLSWithParams(params map[string]string) *TLSAuthProvider {
-	return NewAuthenticationTLS(
-		params["tlsCertFile"],
-		params["tlsKeyFile"],
-	)
+type TLSAuthProvider struct {
+	certificatePath         string
+	privateKeyPath          string
+	allowInsecureConnection bool
 }
 
 // NewAuthenticationTLS initialize the authentication provider
-func NewAuthenticationTLS(certificatePath string, privateKeyPath string) *TLSAuthProvider {
+func NewAuthenticationTLS(certificatePath string, privateKeyPath string,
+	allowInsecureConnection bool) *TLSAuthProvider {
+
 	return &TLSAuthProvider{
-		certificatePath: certificatePath,
-		privateKeyPath:  privateKeyPath,
+		certificatePath:         certificatePath,
+		privateKeyPath:          privateKeyPath,
+		allowInsecureConnection: allowInsecureConnection,
 	}
 }
 
@@ -54,3 +56,40 @@ func (p *TLSAuthProvider) GetTLSCertificate() (*tls.Certificate, error) {
 	cert, err := tls.LoadX509KeyPair(p.certificatePath, p.privateKeyPath)
 	return &cert, err
 }
+
+func (p *TLSAuthProvider) GetTLSConfig(certFile string, allowInsecureConnection bool) (*tls.Config, error) {
+	tlsConfig := &tls.Config{
+		InsecureSkipVerify: allowInsecureConnection,
+	}
+
+	if certFile != "" {
+		caCerts, err := ioutil.ReadFile(certFile)
+		if err != nil {
+			return nil, err
+		}
+
+		tlsConfig.RootCAs = x509.NewCertPool()
+		if !tlsConfig.RootCAs.AppendCertsFromPEM(caCerts) {
+			return nil, errors.New("failed to parse root CAs certificates")
+		}
+	}
+
+	cert, err := p.GetTLSCertificate()
+	if err != nil {
+		return nil, err
+	}
+
+	if cert != nil {
+		tlsConfig.Certificates = []tls.Certificate{*cert}
+	}
+
+	return tlsConfig, nil
+}
+
+func (p *TLSAuthProvider) HasDataForHTTP() bool {
+	return false
+}
+
+func (p *TLSAuthProvider) GetHTTPHeaders() (map[string]string, error) {
+	return nil, errors.New("Unsupported operation")
+}
diff --git a/pkg/auth/token.go b/pkg/auth/token.go
index c928ed585..008b1825b 100644
--- a/pkg/auth/token.go
+++ b/pkg/auth/token.go
@@ -28,18 +28,6 @@ type TokenAuthProvider struct {
 	tokenSupplier func() (string, error)
 }
 
-// NewAuthenticationTokenWithParams return a interface of Provider with string map.
-func NewAuthenticationTokenWithParams(params map[string]string) (*TokenAuthProvider, error) {
-	switch {
-	case params["token"] != "":
-		return NewAuthenticationToken(params["token"]), nil
-	case params["file"] != "":
-		return NewAuthenticationTokenFromFile(params["file"]), nil
-	default:
-		return nil, errors.New("missing configuration for token auth")
-	}
-}
-
 // NewAuthenticationToken return a interface of Provider with a string token.
 func NewAuthenticationToken(token string) *TokenAuthProvider {
 	return &TokenAuthProvider{
@@ -83,3 +71,17 @@ func (p *TokenAuthProvider) GetData() ([]byte, error) {
 	}
 	return []byte(t), nil
 }
+
+func (p *TokenAuthProvider) HasDataForHTTP() bool {
+	return true
+}
+
+func (p *TokenAuthProvider) GetHTTPHeaders() (map[string]string, error) {
+	data, err := p.GetData()
+	if err != nil {
+		return nil, err
+	}
+	headers := make(map[string]string)
+	headers["Authorization"] = "Bearer " + string(data)
+	return headers, nil
+}
diff --git a/pkg/cli/client.go b/pkg/cli/client.go
new file mode 100644
index 000000000..0e201549c
--- /dev/null
+++ b/pkg/cli/client.go
@@ -0,0 +1,381 @@
+// 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 cli
+
+import (
+	"bytes"
+	"encoding/json"
+	"io"
+	"io/ioutil"
+	"net/http"
+	"net/url"
+	"path"
+
+	"github.com/streamnative/pulsarctl/pkg/auth"
+)
+
+// Client is a base client that is used to make http request to the ServiceURL
+type Client struct {
+	ServiceURL   string
+	HTTPClient   *http.Client
+	VersionInfo  string
+	AuthProvider auth.Provider
+}
+
+func (c *Client) newRequest(method, path string) (*request, error) {
+	base, _ := url.Parse(c.ServiceURL)
+	u, err := url.Parse(path)
+	if err != nil {
+		return nil, err
+	}
+
+	req := &request{
+		method: method,
+		url: &url.URL{
+			Scheme: base.Scheme,
+			User:   base.User,
+			Host:   base.Host,
+			Path:   endpoint(base.Path, u.Path),
+		},
+		params: make(url.Values),
+	}
+	return req, nil
+}
+
+func (c *Client) doRequest(r *request) (*http.Response, error) {
+	req, err := r.toHTTP()
+	if err != nil {
+		return nil, err
+	}
+
+	if r.contentType != "" {
+		req.Header.Set("Content-Type", r.contentType)
+	} else {
+		// add default headers
+		req.Header.Set("Content-Type", "application/json")
+	}
+	req.Header.Set("Accept", "application/json")
+	req.Header.Set("User-Agent", c.useragent())
+
+	if c.AuthProvider != nil {
+		if c.AuthProvider.HasDataForHTTP() {
+			headers, err := c.AuthProvider.GetHTTPHeaders()
+			if err != nil {
+				return nil, err
+			}
+			for k, v := range headers {
+				req.Header.Set(k, v)
+			}
+		}
+	}
+
+	hc := c.HTTPClient
+	if hc == nil {
+		hc = http.DefaultClient
+	}
+
+	return hc.Do(req)
+}
+
+// MakeRequest can make a simple request and handle the response by yourself
+func (c *Client) MakeRequest(method, endpoint string) (*http.Response, error) {
+	req, err := c.newRequest(method, endpoint)
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := checkSuccessful(c.doRequest(req))
+	if err != nil {
+		return nil, err
+	}
+
+	return resp, nil
+}
+
+func (c *Client) Get(endpoint string, obj interface{}) error {
+	_, err := c.GetWithQueryParams(endpoint, obj, nil, true)
+	return err
+}
+
+func (c *Client) GetWithQueryParams(endpoint string, obj interface{}, params map[string]string,
+	decode bool) ([]byte, error) {
+
+	req, err := c.newRequest(http.MethodGet, endpoint)
+	if err != nil {
+		return nil, err
+	}
+
+	if params != nil {
+		query := req.url.Query()
+		for k, v := range params {
+			query.Add(k, v)
+		}
+		req.params = query
+	}
+
+	resp, err := checkSuccessful(c.doRequest(req))
+	if err != nil {
+		return nil, err
+	}
+	defer safeRespClose(resp)
+
+	if obj != nil {
+		if err := decodeJSONBody(resp, &obj); err != nil {
+			return nil, err
+		}
+	} else if !decode {
+		body, err := ioutil.ReadAll(resp.Body)
+		if err != nil {
+			return nil, err
+		}
+		return body, err
+	}
+
+	return nil, err
+}
+
+func (c *Client) useragent() string {
+	return c.VersionInfo
+}
+
+func (c *Client) Put(endpoint string, in interface{}) error {
+	return c.PutWithQueryParams(endpoint, in, nil, nil)
+}
+
+func (c *Client) PutWithQueryParams(endpoint string, in, obj interface{}, params map[string]string) error {
+	req, err := c.newRequest(http.MethodPut, endpoint)
+	if err != nil {
+		return err
+	}
+	req.obj = in
+
+	if params != nil {
+		query := req.url.Query()
+		for k, v := range params {
+			query.Add(k, v)
+		}
+		req.params = query
+	}
+
+	resp, err := checkSuccessful(c.doRequest(req))
+	if err != nil {
+		return err
+	}
+	defer safeRespClose(resp)
+
+	if obj != nil {
+		if err := decodeJSONBody(resp, &obj); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (c *Client) PutWithMultiPart(endpoint string, body io.Reader, contentType string) error {
+	req, err := c.newRequest(http.MethodPut, endpoint)
+	if err != nil {
+		return err
+	}
+	req.body = body
+	req.contentType = contentType
+
+	// nolint
+	resp, err := checkSuccessful(c.doRequest(req))
+	if err != nil {
+		return err
+	}
+	defer safeRespClose(resp)
+
+	return nil
+}
+
+func (c *Client) Delete(endpoint string) error {
+	return c.DeleteWithQueryParams(endpoint, nil)
+}
+
+func (c *Client) DeleteWithQueryParams(endpoint string, params map[string]string) error {
+	req, err := c.newRequest(http.MethodDelete, endpoint)
+	if err != nil {
+		return err
+	}
+
+	if params != nil {
+		query := req.url.Query()
+		for k, v := range params {
+			query.Add(k, v)
+		}
+		req.params = query
+	}
+
+	// nolint
+	resp, err := checkSuccessful(c.doRequest(req))
+	if err != nil {
+		return err
+	}
+	defer safeRespClose(resp)
+
+	return nil
+}
+
+func (c *Client) Post(endpoint string, in interface{}) error {
+	return c.PostWithObj(endpoint, in, nil)
+}
+
+func (c *Client) PostWithObj(endpoint string, in, obj interface{}) error {
+	req, err := c.newRequest(http.MethodPost, endpoint)
+	if err != nil {
+		return err
+	}
+	req.obj = in
+
+	// nolint
+	resp, err := checkSuccessful(c.doRequest(req))
+	if err != nil {
+		return err
+	}
+	defer safeRespClose(resp)
+	if obj != nil {
+		if err := decodeJSONBody(resp, &obj); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (c *Client) PostWithMultiPart(endpoint string, in interface{}, body io.Reader, contentType string) error {
+	req, err := c.newRequest(http.MethodPost, endpoint)
+	if err != nil {
+		return err
+	}
+	req.obj = in
+	req.body = body
+	req.contentType = contentType
+
+	// nolint
+	resp, err := checkSuccessful(c.doRequest(req))
+	if err != nil {
+		return err
+	}
+	defer safeRespClose(resp)
+
+	return nil
+}
+
+type request struct {
+	method      string
+	contentType string
+	url         *url.URL
+	params      url.Values
+
+	obj  interface{}
+	body io.Reader
+}
+
+func (r *request) toHTTP() (*http.Request, error) {
+	r.url.RawQuery = r.params.Encode()
+
+	// add a request body if there is one
+	if r.body == nil && r.obj != nil {
+		body, err := encodeJSONBody(r.obj)
+		if err != nil {
+			return nil, err
+		}
+		r.body = body
+	}
+
+	req, err := http.NewRequest(r.method, r.url.RequestURI(), r.body)
+	if err != nil {
+		return nil, err
+	}
+
+	req.URL.Host = r.url.Host
+	req.URL.Scheme = r.url.Scheme
+	req.Host = r.url.Host
+	return req, nil
+}
+
+// respIsOk is used to validate a successful http status code
+func respIsOk(resp *http.Response) bool {
+	return resp.StatusCode >= http.StatusOK && resp.StatusCode <= http.StatusNoContent
+}
+
+// checkSuccessful checks for a valid response and parses an error
+func checkSuccessful(resp *http.Response, err error) (*http.Response, error) {
+	if err != nil {
+		safeRespClose(resp)
+		return nil, err
+	}
+
+	if !respIsOk(resp) {
+		defer safeRespClose(resp)
+		return nil, responseError(resp)
+	}
+
+	return resp, nil
+}
+
+func endpoint(parts ...string) string {
+	return path.Join(parts...)
+}
+
+// encodeJSONBody is used to JSON encode a body
+func encodeJSONBody(obj interface{}) (io.Reader, error) {
+	buf := bytes.NewBuffer(nil)
+	enc := json.NewEncoder(buf)
+	if err := enc.Encode(obj); err != nil {
+		return nil, err
+	}
+	return buf, nil
+}
+
+// decodeJSONBody is used to JSON decode a body
+func decodeJSONBody(resp *http.Response, out interface{}) error {
+	dec := json.NewDecoder(resp.Body)
+	return dec.Decode(out)
+}
+
+// safeRespClose is used to close a response body
+func safeRespClose(resp *http.Response) {
+	if resp != nil {
+		// ignore error since it is closing a response body
+		_ = resp.Body.Close()
+	}
+}
+
+// responseError is used to parse a response into a pulsar error
+func responseError(resp *http.Response) error {
+	var e Error
+	body, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		e.Reason = err.Error()
+		e.Code = resp.StatusCode
+		return e
+	}
+
+	json.Unmarshal(body, &e)
+
+	e.Code = resp.StatusCode
+
+	if e.Reason == "" {
+		e.Reason = unknownErrorReason
+	}
+
+	return e
+}
diff --git a/pkg/cli/errors.go b/pkg/cli/errors.go
new file mode 100644
index 000000000..3a7e91a4c
--- /dev/null
+++ b/pkg/cli/errors.go
@@ -0,0 +1,37 @@
+// 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 cli
+
+import "fmt"
+
+const unknownErrorReason = "Unknown pulsar error"
+
+// Error is a admin error type
+type Error struct {
+	Reason string `json:"reason"`
+	Code   int
+}
+
+func (e Error) Error() string {
+	return fmt.Sprintf("code: %d reason: %s", e.Code, e.Reason)
+}
+
+func IsAdminError(err error) bool {
+	_, ok := err.(Error)
+	return ok
+}
diff --git a/pkg/cmdutils/config.go b/pkg/cmdutils/config.go
index 3c1bc277f..72aaf8feb 100644
--- a/pkg/cmdutils/config.go
+++ b/pkg/cmdutils/config.go
@@ -21,7 +21,6 @@ import (
 	"log"
 	"os"
 
-	"github.com/streamnative/pulsarctl/pkg/auth"
 	"github.com/streamnative/pulsarctl/pkg/pulsar"
 	"github.com/streamnative/pulsarctl/pkg/pulsar/common"
 
@@ -100,16 +99,12 @@ func (c *ClusterConfig) Client(version common.APIVersion) pulsar.Client {
 		config.WebServiceURL = c.WebServiceURL
 	}
 
-	if len(c.TLSTrustCertsFilePath) > 0 && c.TLSTrustCertsFilePath != config.TLSOptions.TrustCertsFilePath {
-		config.TLSOptions.TrustCertsFilePath = c.TLSTrustCertsFilePath
+	if len(c.TLSTrustCertsFilePath) > 0 && c.TLSTrustCertsFilePath != config.TLSCertFile {
+		config.TLSCertFile = c.TLSTrustCertsFilePath
 	}
 
 	if c.TLSAllowInsecureConnection {
-		config.TLSOptions.AllowInsecureConnection = true
-	}
-
-	if len(c.AuthParams) > 0 && c.AuthParams != config.AuthParams {
-		config.AuthParams = c.AuthParams
+		config.TLSAllowInsecureConnection = true
 	}
 
 	if len(c.Token) > 0 && len(c.TokenFile) > 0 {
@@ -122,23 +117,8 @@ func (c *ClusterConfig) Client(version common.APIVersion) pulsar.Client {
 			logger.Critical("the token and tls can not be specified at the same time")
 			os.Exit(1)
 		}
-
-		tokenParams := make(map[string]string)
-		if len(c.Token) > 0 {
-			tokenParams["token"] = c.Token
-		}
-
-		if len(c.TokenFile) > 0 {
-			tokenParams["file"] = c.TokenFile
-		}
-
-		tokenAuth, err := auth.NewAuthenticationTokenWithParams(tokenParams)
-		if err != nil {
-			logger.Critical("%s\n", err.Error())
-			os.Exit(1)
-		}
-
-		config.TokenAuth = tokenAuth
+		config.TokenFile = c.TokenFile
+		config.Token = c.Token
 	}
 
 	config.APIVersion = version
diff --git a/pkg/ctl/sources/get_test.go b/pkg/ctl/sources/get_test.go
index ccae2a88c..ef3f306aa 100644
--- a/pkg/ctl/sources/get_test.go
+++ b/pkg/ctl/sources/get_test.go
@@ -89,6 +89,6 @@ func TestGetFailureSource(t *testing.T) {
 	}
 	getOut, execErr, _ := TestSourcesCommands(getSourcesCmd, failureGetArgs)
 	assert.NotNil(t, execErr)
-	exceptedErr := "error: Source test-source-get doesn't exist\n"
+	exceptedErr := "error: code: 404 reason: Source test-source-get doesn't exist\n"
 	assert.Equal(t, getOut.String(), exceptedErr)
 }
diff --git a/pkg/pulsar/admin.go b/pkg/pulsar/admin.go
index 3d0b92c96..1da07b8dd 100644
--- a/pkg/pulsar/admin.go
+++ b/pkg/pulsar/admin.go
@@ -18,64 +18,21 @@
 package pulsar
 
 import (
-	"bytes"
-	"crypto/tls"
-	"crypto/x509"
-	"encoding/json"
-	"errors"
-	"io"
-	"io/ioutil"
 	"net/http"
-	"net/url"
 	"path"
 	"strings"
-	"time"
 
 	"github.com/streamnative/pulsarctl/pkg/auth"
+	"github.com/streamnative/pulsarctl/pkg/cli"
 	"github.com/streamnative/pulsarctl/pkg/pulsar/common"
 	"github.com/streamnative/pulsarctl/pkg/pulsar/utils"
 )
 
-const (
-	DefaultWebServiceURL       = "http://localhost:8080"
-	DefaultHTTPTimeOutDuration = 5 * time.Minute
-)
-
-var ReleaseVersion = "None"
-
-// Config is used to configure the admin client
-type Config struct {
-	WebServiceURL string
-	HTTPTimeout   time.Duration
-	HTTPClient    *http.Client
-	APIVersion    common.APIVersion
-
-	Auth       *auth.TLSAuthProvider
-	AuthParams string
-	TLSOptions *TLSOptions
-	TokenAuth  *auth.TokenAuthProvider
-}
-
 type TLSOptions struct {
 	TrustCertsFilePath      string
 	AllowInsecureConnection bool
 }
 
-// DefaultConfig returns a default configuration for the pulsar admin client
-func DefaultConfig() *Config {
-	config := &Config{
-		WebServiceURL: DefaultWebServiceURL,
-		HTTPClient: &http.Client{
-			Timeout: DefaultHTTPTimeOutDuration,
-		},
-
-		TLSOptions: &TLSOptions{
-			AllowInsecureConnection: false,
-		},
-	}
-	return config
-}
-
 // Client provides a client to the Pulsar Restful API
 type Client interface {
 	Clusters() Clusters
@@ -94,19 +51,9 @@ type Client interface {
 	FunctionsWorker() FunctionsWorker
 }
 
-type client struct {
-	webServiceURL string
-	apiVersion    string
-	httpClient    *http.Client
-	versionInfo   string
-
-	// TLS config
-	auth       *auth.TLSAuthProvider
-	authParams string
-	tlsOptions *TLSOptions
-	transport  *http.Transport
-
-	tokenAuth *auth.TokenAuthProvider
+type pulsarClient struct {
+	Client     *cli.Client
+	APIVersion common.APIVersion
 }
 
 // New returns a new client
@@ -115,405 +62,65 @@ func New(config *Config) (Client, error) {
 		config.WebServiceURL = DefaultWebServiceURL
 	}
 
-	c := &client{
-		apiVersion:    config.APIVersion.String(),
-		webServiceURL: config.WebServiceURL,
-		versionInfo:   ReleaseVersion,
-		tokenAuth:     config.TokenAuth,
-	}
-
-	if strings.HasPrefix(c.webServiceURL, "https://") {
-		c.authParams = config.AuthParams
-		c.tlsOptions = config.TLSOptions
-		mapAuthParams := make(map[string]string)
-
-		err := json.Unmarshal([]byte(c.authParams), &mapAuthParams)
-		if err != nil {
-			return nil, err
-		}
-		c.auth = auth.NewAuthenticationTLSWithParams(mapAuthParams)
-
-		tlsConf, err := c.getTLSConfig()
-		if err != nil {
-			return nil, err
-		}
-
-		c.transport = &http.Transport{
-			TLSHandshakeTimeout: 15 * time.Second,
-			MaxIdleConnsPerHost: 10,
-			TLSClientConfig:     tlsConf,
-		}
-	}
-
-	return c, nil
-}
-
-func (c *client) getTLSConfig() (*tls.Config, error) {
-	tlsConfig := &tls.Config{
-		InsecureSkipVerify: c.tlsOptions.AllowInsecureConnection,
-	}
-
-	if c.tlsOptions.TrustCertsFilePath != "" {
-		caCerts, err := ioutil.ReadFile(c.tlsOptions.TrustCertsFilePath)
-		if err != nil {
-			return nil, err
-		}
-
-		tlsConfig.RootCAs = x509.NewCertPool()
-		ok := tlsConfig.RootCAs.AppendCertsFromPEM(caCerts)
-		if !ok {
-			return nil, errors.New("failed to parse root CAs certificates")
-		}
+	c := &pulsarClient{
+		APIVersion: config.APIVersion,
+		Client: &cli.Client{
+			ServiceURL:  config.WebServiceURL,
+			VersionInfo: ReleaseVersion,
+			HTTPClient: &http.Client{
+				Timeout: DefaultHTTPTimeOutDuration,
+			},
+		},
 	}
 
-	cert, err := c.auth.GetTLSCertificate()
+	err := c.initAuth(config)
 	if err != nil {
 		return nil, err
 	}
 
-	if cert != nil {
-		tlsConfig.Certificates = []tls.Certificate{*cert}
-	}
-
-	return tlsConfig, nil
-}
-
-func (c *client) endpoint(componentPath string, parts ...string) string {
-	return path.Join(utils.MakeHTTPPath(c.apiVersion, componentPath), endpoint(parts...))
+	return c, nil
 }
 
-// get is used to do a GET request against an endpoint
-// and deserialize the response into an interface
-
-func (c *client) getWithQueryParams(endpoint string, obj interface{}, params map[string]string,
-	decode bool) ([]byte, error) {
-
-	req, err := c.newRequest(http.MethodGet, endpoint)
-	if err != nil {
-		return nil, err
-	}
-
-	if params != nil {
-		query := req.url.Query()
-		for k, v := range params {
-			query.Add(k, v)
-		}
-		req.params = query
-	}
-
-	resp, err := checkSuccessful(c.doRequest(req))
-	if err != nil {
-		return nil, err
-	}
-	defer safeRespClose(resp)
-
-	if obj != nil {
-		if err := decodeJSONBody(resp, &obj); err != nil {
-			return nil, err
-		}
-	} else if !decode {
-		body, err := ioutil.ReadAll(resp.Body)
+func (c *pulsarClient) initAuth(config *Config) error {
+	if strings.HasPrefix(config.WebServiceURL, "https") {
+		err := c.initTLS(config)
 		if err != nil {
-			return nil, err
-		}
-		return body, err
-	}
-
-	return nil, err
-}
-
-func (c *client) get(endpoint string, obj interface{}) error {
-	_, err := c.getWithQueryParams(endpoint, obj, nil, true)
-	return err
-}
-
-func (c *client) put(endpoint string, in interface{}) error {
-	return c.putWithQueryParams(endpoint, in, nil, nil)
-}
-
-func (c *client) putWithQueryParams(endpoint string, in, obj interface{}, params map[string]string) error {
-	req, err := c.newRequest(http.MethodPut, endpoint)
-	if err != nil {
-		return err
-	}
-	req.obj = in
-
-	if params != nil {
-		query := req.url.Query()
-		for k, v := range params {
-			query.Add(k, v)
-		}
-		req.params = query
-	}
-
-	resp, err := checkSuccessful(c.doRequest(req))
-	if err != nil {
-		return err
-	}
-	defer safeRespClose(resp)
-
-	if obj != nil {
-		if err := decodeJSONBody(resp, &obj); err != nil {
 			return err
 		}
 	}
 
-	return nil
-}
-
-func (c *client) delete(endpoint string) error {
-	return c.deleteWithQueryParams(endpoint, nil, nil)
-}
-
-func (c *client) deleteWithQueryParams(endpoint string, obj interface{}, params map[string]string) error {
-	req, err := c.newRequest(http.MethodDelete, endpoint)
-	if err != nil {
-		return err
-	}
-
-	if params != nil {
-		query := req.url.Query()
-		for k, v := range params {
-			query.Add(k, v)
-		}
-		req.params = query
-	}
-
-	resp, err := checkSuccessful(c.doRequest(req))
-	if err != nil {
-		return err
-	}
-	defer safeRespClose(resp)
-
-	if obj != nil {
-		if err := decodeJSONBody(resp, &obj); err != nil {
-			return err
-		}
+	if config.TokenFile != "" || config.Token != "" {
+		c.initToken(config)
 	}
 
 	return nil
 }
 
-func (c *client) post(endpoint string, in interface{}) error {
-	return c.postWithObj(endpoint, in, nil)
-}
-
-func (c *client) postWithObj(endpoint string, in, obj interface{}) error {
-	req, err := c.newRequest(http.MethodPost, endpoint)
-	if err != nil {
-		return err
-	}
-	req.obj = in
-
-	// nolint
-	resp, err := checkSuccessful(c.doRequest(req))
-	if err != nil {
-		return err
-	}
-	defer safeRespClose(resp)
-	if obj != nil {
-		if err := decodeJSONBody(resp, &obj); err != nil {
-			return err
-		}
-	}
-
-	return nil
-}
-
-func (c *client) putWithMultiPart(endpoint string, body io.Reader, contentType string) error {
-	req, err := c.newRequest(http.MethodPut, endpoint)
-	if err != nil {
-		return err
-	}
-	req.body = body
-	req.contentType = contentType
-
-	// nolint
-	resp, err := checkSuccessful(c.doRequest(req))
-	if err != nil {
-		return err
-	}
-	defer safeRespClose(resp)
-
-	return nil
-}
-
-func (c *client) postWithMultiPart(endpoint string, in interface{}, body io.Reader, contentType string) error {
-	req, err := c.newRequest(http.MethodPost, endpoint)
+func (c *pulsarClient) initTLS(config *Config) error {
+	tlsAuth := auth.NewAuthenticationTLS(config.TLSCertFile, config.TLSKeyFile, config.TLSAllowInsecureConnection)
+	tlsConf, err := tlsAuth.GetTLSConfig(config.TLSCertFile, config.TLSAllowInsecureConnection)
 	if err != nil {
 		return err
 	}
-	req.obj = in
-	req.body = body
-	req.contentType = contentType
 
-	// nolint
-	resp, err := checkSuccessful(c.doRequest(req))
-	if err != nil {
-		return err
+	c.Client.HTTPClient.Transport = &http.Transport{
+		MaxIdleConnsPerHost: 10,
+		TLSClientConfig:     tlsConf,
 	}
-	defer safeRespClose(resp)
 
 	return nil
 }
 
-type request struct {
-	method      string
-	contentType string
-	url         *url.URL
-	params      url.Values
-
-	obj  interface{}
-	body io.Reader
-}
-
-func (r *request) toHTTP() (*http.Request, error) {
-	r.url.RawQuery = r.params.Encode()
-
-	// add a request body if there is one
-	if r.body == nil && r.obj != nil {
-		body, err := encodeJSONBody(r.obj)
-		if err != nil {
-			return nil, err
-		}
-		r.body = body
-	}
-
-	req, err := http.NewRequest(r.method, r.url.RequestURI(), r.body)
-	if err != nil {
-		return nil, err
-	}
-
-	req.URL.Host = r.url.Host
-	req.URL.Scheme = r.url.Scheme
-	req.Host = r.url.Host
-	return req, nil
-}
-
-func (c *client) newRequest(method, path string) (*request, error) {
-	base, _ := url.Parse(c.webServiceURL)
-	u, err := url.Parse(path)
-	if err != nil {
-		return nil, err
-	}
-
-	req := &request{
-		method: method,
-		url: &url.URL{
-			Scheme: base.Scheme,
-			User:   base.User,
-			Host:   base.Host,
-			Path:   endpoint(base.Path, u.Path),
-		},
-		params: make(url.Values),
-	}
-	return req, nil
-}
-
-func (c *client) useragent() string {
-	return c.versionInfo
-}
-
-func (c *client) doRequest(r *request) (*http.Response, error) {
-	req, err := r.toHTTP()
-	if err != nil {
-		return nil, err
-	}
-
-	if r.contentType != "" {
-		req.Header.Set("Content-Type", r.contentType)
-	} else {
-		// add default headers
-		req.Header.Set("Content-Type", "application/json")
-		req.Header.Set("Accept", "application/json")
-	}
-
-	if c.tokenAuth != nil {
-		data, _ := c.tokenAuth.GetData()
-		req.Header.Set("Authorization", "Bearer "+string(data))
-	}
-
-	req.Header.Set("User-Agent", c.useragent())
-
-	hc := c.httpClient
-	if hc == nil {
-		hc = &http.Client{
-			Timeout: DefaultHTTPTimeOutDuration,
-		}
-	}
-
-	if c.transport != nil {
-		hc.Transport = c.transport
-	}
-
-	return hc.Do(req)
-}
-
-// encodeJSONBody is used to JSON encode a body
-func encodeJSONBody(obj interface{}) (io.Reader, error) {
-	buf := bytes.NewBuffer(nil)
-	enc := json.NewEncoder(buf)
-	if err := enc.Encode(obj); err != nil {
-		return nil, err
-	}
-	return buf, nil
-}
-
-// decodeJSONBody is used to JSON decode a body
-func decodeJSONBody(resp *http.Response, out interface{}) error {
-	dec := json.NewDecoder(resp.Body)
-	return dec.Decode(out)
-}
-
-// safeRespClose is used to close a response body
-func safeRespClose(resp *http.Response) {
-	if resp != nil {
-		// ignore error since it is closing a response body
-		_ = resp.Body.Close()
+func (c *pulsarClient) initToken(config *Config) {
+	if config.Token != "" {
+		c.Client.AuthProvider = auth.NewAuthenticationToken(config.Token)
 	}
-}
-
-// responseError is used to parse a response into a pulsar error
-func responseError(resp *http.Response) error {
-	var e common.Error
-	body, err := ioutil.ReadAll(resp.Body)
-	if err != nil {
-		e.Reason = err.Error()
-		e.Code = resp.StatusCode
-		return e
-	}
-
-	json.Unmarshal(body, &e)
-
-	e.Code = resp.StatusCode
 
-	if e.Reason == "" {
-		e.Reason = common.UnknownErrorReason
+	if config.TokenFile != "" {
+		c.Client.AuthProvider = auth.NewAuthenticationTokenFromFile(config.TokenFile)
 	}
-
-	return e
-}
-
-// respIsOk is used to validate a successful http status code
-func respIsOk(resp *http.Response) bool {
-	return resp.StatusCode >= http.StatusOK && resp.StatusCode <= http.StatusNoContent
-}
-
-// checkSuccessful checks for a valid response and parses an error
-func checkSuccessful(resp *http.Response, err error) (*http.Response, error) {
-	if err != nil {
-		safeRespClose(resp)
-		return nil, err
-	}
-
-	if !respIsOk(resp) {
-		defer safeRespClose(resp)
-		return nil, responseError(resp)
-	}
-
-	return resp, nil
 }
 
-func endpoint(parts ...string) string {
-	return path.Join(parts...)
+func (c *pulsarClient) endpoint(componentPath string, parts ...string) string {
+	return path.Join(utils.MakeHTTPPath(c.APIVersion.String(), componentPath), path.Join(parts...))
 }
diff --git a/pkg/pulsar/admin_config.go b/pkg/pulsar/admin_config.go
new file mode 100644
index 000000000..9c6706d10
--- /dev/null
+++ b/pkg/pulsar/admin_config.go
@@ -0,0 +1,56 @@
+// 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 pulsar
+
+import (
+	"time"
+
+	"github.com/streamnative/pulsarctl/pkg/pulsar/common"
+)
+
+const (
+	DefaultWebServiceURL       = "http://localhost:8080"
+	DefaultHTTPTimeOutDuration = 5 * time.Minute
+)
+
+var ReleaseVersion = "None"
+
+// Config is used to configure the admin client
+type Config struct {
+	WebServiceURL string
+	HTTPTimeout   time.Duration
+	// TODO: api version should apply to the method
+	APIVersion common.APIVersion
+
+	//Auth       *auth.TLSAuthProvider
+	TLSCertFile                string
+	TLSKeyFile                 string
+	TLSAllowInsecureConnection bool
+
+	// Token and TokenFile is used to config the pulsarctl using token to authentication
+	Token     string
+	TokenFile string
+}
+
+// DefaultConfig returns a default configuration for the pulsar admin client
+func DefaultConfig() *Config {
+	config := &Config{
+		WebServiceURL: DefaultWebServiceURL,
+	}
+	return config
+}
diff --git a/pkg/pulsar/broker_stats.go b/pkg/pulsar/broker_stats.go
index 183492bc5..525810375 100644
--- a/pkg/pulsar/broker_stats.go
+++ b/pkg/pulsar/broker_stats.go
@@ -40,22 +40,22 @@ type BrokerStats interface {
 }
 
 type brokerStats struct {
-	client   *client
+	pulsar   *pulsarClient
 	basePath string
 }
 
 // BrokerStats is used to access the broker stats endpoints
-func (c *client) BrokerStats() BrokerStats {
+func (c *pulsarClient) BrokerStats() BrokerStats {
 	return &brokerStats{
-		client:   c,
+		pulsar:   c,
 		basePath: "/broker-stats",
 	}
 }
 
 func (bs *brokerStats) GetMetrics() ([]utils.Metrics, error) {
-	endpoint := bs.client.endpoint(bs.basePath, "/metrics")
+	endpoint := bs.pulsar.endpoint(bs.basePath, "/metrics")
 	var response []utils.Metrics
-	err := bs.client.get(endpoint, &response)
+	err := bs.pulsar.Client.Get(endpoint, &response)
 	if err != nil {
 		return nil, err
 	}
@@ -64,9 +64,9 @@ func (bs *brokerStats) GetMetrics() ([]utils.Metrics, error) {
 }
 
 func (bs *brokerStats) GetMBeans() ([]utils.Metrics, error) {
-	endpoint := bs.client.endpoint(bs.basePath, "/mbeans")
+	endpoint := bs.pulsar.endpoint(bs.basePath, "/mbeans")
 	var response []utils.Metrics
-	err := bs.client.get(endpoint, &response)
+	err := bs.pulsar.Client.Get(endpoint, &response)
 	if err != nil {
 		return nil, err
 	}
@@ -75,8 +75,8 @@ func (bs *brokerStats) GetMBeans() ([]utils.Metrics, error) {
 }
 
 func (bs *brokerStats) GetTopics() (string, error) {
-	endpoint := bs.client.endpoint(bs.basePath, "/topics")
-	buf, err := bs.client.getWithQueryParams(endpoint, nil, nil, false)
+	endpoint := bs.pulsar.endpoint(bs.basePath, "/topics")
+	buf, err := bs.pulsar.Client.GetWithQueryParams(endpoint, nil, nil, false)
 	if err != nil {
 		return "", err
 	}
@@ -85,9 +85,9 @@ func (bs *brokerStats) GetTopics() (string, error) {
 }
 
 func (bs *brokerStats) GetLoadReport() (*utils.LocalBrokerData, error) {
-	endpoint := bs.client.endpoint(bs.basePath, "/load-report")
+	endpoint := bs.pulsar.endpoint(bs.basePath, "/load-report")
 	response := utils.NewLocalBrokerData()
-	err := bs.client.get(endpoint, &response)
+	err := bs.pulsar.Client.Get(endpoint, &response)
 	if err != nil {
 		return nil, nil
 	}
@@ -95,9 +95,9 @@ func (bs *brokerStats) GetLoadReport() (*utils.LocalBrokerData, error) {
 }
 
 func (bs *brokerStats) GetAllocatorStats(allocatorName string) (*utils.AllocatorStats, error) {
-	endpoint := bs.client.endpoint(bs.basePath, "/allocator-stats", allocatorName)
+	endpoint := bs.pulsar.endpoint(bs.basePath, "/allocator-stats", allocatorName)
 	var allocatorStats utils.AllocatorStats
-	err := bs.client.get(endpoint, &allocatorStats)
+	err := bs.pulsar.Client.Get(endpoint, &allocatorStats)
 	if err != nil {
 		return nil, err
 	}
diff --git a/pkg/pulsar/brokers.go b/pkg/pulsar/brokers.go
index ecc885303..2a96890c0 100644
--- a/pkg/pulsar/brokers.go
+++ b/pkg/pulsar/brokers.go
@@ -58,22 +58,22 @@ type Brokers interface {
 }
 
 type broker struct {
-	client   *client
+	pulsar   *pulsarClient
 	basePath string
 }
 
 // Brokers is used to access the brokers endpoints
-func (c *client) Brokers() Brokers {
+func (c *pulsarClient) Brokers() Brokers {
 	return &broker{
-		client:   c,
+		pulsar:   c,
 		basePath: "/brokers",
 	}
 }
 
 func (b *broker) GetActiveBrokers(cluster string) ([]string, error) {
-	endpoint := b.client.endpoint(b.basePath, cluster)
+	endpoint := b.pulsar.endpoint(b.basePath, cluster)
 	var res []string
-	err := b.client.get(endpoint, &res)
+	err := b.pulsar.Client.Get(endpoint, &res)
 	if err != nil {
 		return nil, err
 	}
@@ -81,9 +81,9 @@ func (b *broker) GetActiveBrokers(cluster string) ([]string, error) {
 }
 
 func (b *broker) GetDynamicConfigurationNames() ([]string, error) {
-	endpoint := b.client.endpoint(b.basePath, "/configuration/")
+	endpoint := b.pulsar.endpoint(b.basePath, "/configuration/")
 	var res []string
-	err := b.client.get(endpoint, &res)
+	err := b.pulsar.Client.Get(endpoint, &res)
 	if err != nil {
 		return nil, err
 	}
@@ -91,9 +91,9 @@ func (b *broker) GetDynamicConfigurationNames() ([]string, error) {
 }
 
 func (b *broker) GetOwnedNamespaces(cluster, brokerURL string) (map[string]utils.NamespaceOwnershipStatus, error) {
-	endpoint := b.client.endpoint(b.basePath, cluster, brokerURL, "ownedNamespaces")
+	endpoint := b.pulsar.endpoint(b.basePath, cluster, brokerURL, "ownedNamespaces")
 	var res map[string]utils.NamespaceOwnershipStatus
-	err := b.client.get(endpoint, &res)
+	err := b.pulsar.Client.Get(endpoint, &res)
 	if err != nil {
 		return nil, err
 	}
@@ -102,19 +102,19 @@ func (b *broker) GetOwnedNamespaces(cluster, brokerURL string) (map[string]utils
 
 func (b *broker) UpdateDynamicConfiguration(configName, configValue string) error {
 	value := url.QueryEscape(configValue)
-	endpoint := b.client.endpoint(b.basePath, "/configuration/", configName, value)
-	return b.client.post(endpoint, nil)
+	endpoint := b.pulsar.endpoint(b.basePath, "/configuration/", configName, value)
+	return b.pulsar.Client.Post(endpoint, nil)
 }
 
 func (b *broker) DeleteDynamicConfiguration(configName string) error {
-	endpoint := b.client.endpoint(b.basePath, "/configuration/", configName)
-	return b.client.delete(endpoint)
+	endpoint := b.pulsar.endpoint(b.basePath, "/configuration/", configName)
+	return b.pulsar.Client.Delete(endpoint)
 }
 
 func (b *broker) GetRuntimeConfigurations() (map[string]string, error) {
-	endpoint := b.client.endpoint(b.basePath, "/configuration/", "runtime")
+	endpoint := b.pulsar.endpoint(b.basePath, "/configuration/", "runtime")
 	var res map[string]string
-	err := b.client.get(endpoint, &res)
+	err := b.pulsar.Client.Get(endpoint, &res)
 	if err != nil {
 		return nil, err
 	}
@@ -122,9 +122,9 @@ func (b *broker) GetRuntimeConfigurations() (map[string]string, error) {
 }
 
 func (b *broker) GetInternalConfigurationData() (*utils.InternalConfigurationData, error) {
-	endpoint := b.client.endpoint(b.basePath, "/internal-configuration")
+	endpoint := b.pulsar.endpoint(b.basePath, "/internal-configuration")
 	var res utils.InternalConfigurationData
-	err := b.client.get(endpoint, &res)
+	err := b.pulsar.Client.Get(endpoint, &res)
 	if err != nil {
 		return nil, err
 	}
@@ -132,9 +132,9 @@ func (b *broker) GetInternalConfigurationData() (*utils.InternalConfigurationDat
 }
 
 func (b *broker) GetAllDynamicConfigurations() (map[string]string, error) {
-	endpoint := b.client.endpoint(b.basePath, "/configuration/", "values")
+	endpoint := b.pulsar.endpoint(b.basePath, "/configuration/", "values")
 	var res map[string]string
-	err := b.client.get(endpoint, &res)
+	err := b.pulsar.Client.Get(endpoint, &res)
 	if err != nil {
 		return nil, err
 	}
@@ -142,9 +142,9 @@ func (b *broker) GetAllDynamicConfigurations() (map[string]string, error) {
 }
 
 func (b *broker) HealthCheck() error {
-	endpoint := b.client.endpoint(b.basePath, "/health")
+	endpoint := b.pulsar.endpoint(b.basePath, "/health")
 
-	buf, err := b.client.getWithQueryParams(endpoint, nil, nil, false)
+	buf, err := b.pulsar.Client.GetWithQueryParams(endpoint, nil, nil, false)
 	if err != nil {
 		return err
 	}
diff --git a/pkg/pulsar/cluster.go b/pkg/pulsar/cluster.go
index 6c215cf9a..15e411dd2 100644
--- a/pkg/pulsar/cluster.go
+++ b/pkg/pulsar/cluster.go
@@ -61,82 +61,82 @@ type Clusters interface {
 }
 
 type clusters struct {
-	client   *client
+	pulsar   *pulsarClient
 	basePath string
 }
 
 // Clusters is used to access the cluster endpoints.
-func (c *client) Clusters() Clusters {
+func (c *pulsarClient) Clusters() Clusters {
 	return &clusters{
-		client:   c,
+		pulsar:   c,
 		basePath: "/clusters",
 	}
 }
 
 func (c *clusters) List() ([]string, error) {
 	var clusters []string
-	err := c.client.get(c.client.endpoint(c.basePath), &clusters)
+	err := c.pulsar.Client.Get(c.pulsar.endpoint(c.basePath), &clusters)
 	return clusters, err
 }
 
 func (c *clusters) Get(name string) (utils.ClusterData, error) {
 	cdata := utils.ClusterData{}
-	endpoint := c.client.endpoint(c.basePath, name)
-	err := c.client.get(endpoint, &cdata)
+	endpoint := c.pulsar.endpoint(c.basePath, name)
+	err := c.pulsar.Client.Get(endpoint, &cdata)
 	return cdata, err
 }
 
 func (c *clusters) Create(cdata utils.ClusterData) error {
-	endpoint := c.client.endpoint(c.basePath, cdata.Name)
-	return c.client.put(endpoint, &cdata)
+	endpoint := c.pulsar.endpoint(c.basePath, cdata.Name)
+	return c.pulsar.Client.Put(endpoint, &cdata)
 }
 
 func (c *clusters) Delete(name string) error {
-	endpoint := c.client.endpoint(c.basePath, name)
-	return c.client.delete(endpoint)
+	endpoint := c.pulsar.endpoint(c.basePath, name)
+	return c.pulsar.Client.Delete(endpoint)
 }
 
 func (c *clusters) Update(cdata utils.ClusterData) error {
-	endpoint := c.client.endpoint(c.basePath, cdata.Name)
-	return c.client.post(endpoint, &cdata)
+	endpoint := c.pulsar.endpoint(c.basePath, cdata.Name)
+	return c.pulsar.Client.Post(endpoint, &cdata)
 }
 
 func (c *clusters) GetPeerClusters(name string) ([]string, error) {
 	var peerClusters []string
-	endpoint := c.client.endpoint(c.basePath, name, "peers")
-	err := c.client.get(endpoint, &peerClusters)
+	endpoint := c.pulsar.endpoint(c.basePath, name, "peers")
+	err := c.pulsar.Client.Get(endpoint, &peerClusters)
 	return peerClusters, err
 }
 
 func (c *clusters) UpdatePeerClusters(cluster string, peerClusters []string) error {
-	endpoint := c.client.endpoint(c.basePath, cluster, "peers")
-	return c.client.post(endpoint, peerClusters)
+	endpoint := c.pulsar.endpoint(c.basePath, cluster, "peers")
+	return c.pulsar.Client.Post(endpoint, peerClusters)
 }
 
 func (c *clusters) CreateFailureDomain(data utils.FailureDomainData) error {
-	endpoint := c.client.endpoint(c.basePath, data.ClusterName, "failureDomains", data.DomainName)
-	return c.client.post(endpoint, &data)
+	endpoint := c.pulsar.endpoint(c.basePath, data.ClusterName, "failureDomains", data.DomainName)
+	return c.pulsar.Client.Post(endpoint, &data)
 }
 
 func (c *clusters) GetFailureDomain(clusterName string, domainName string) (utils.FailureDomainData, error) {
 	var res utils.FailureDomainData
-	endpoint := c.client.endpoint(c.basePath, clusterName, "failureDomains", domainName)
-	err := c.client.get(endpoint, &res)
+	endpoint := c.pulsar.endpoint(c.basePath, clusterName, "failureDomains", domainName)
+	err := c.pulsar.Client.Get(endpoint, &res)
 	return res, err
 }
 
 func (c *clusters) ListFailureDomains(clusterName string) (utils.FailureDomainMap, error) {
 	var domainData utils.FailureDomainMap
-	endpoint := c.client.endpoint(c.basePath, clusterName, "failureDomains")
-	err := c.client.get(endpoint, &domainData)
+	endpoint := c.pulsar.endpoint(c.basePath, clusterName, "failureDomains")
+	err := c.pulsar.Client.Get(endpoint, &domainData)
 	return domainData, err
 }
 
 func (c *clusters) DeleteFailureDomain(data utils.FailureDomainData) error {
-	endpoint := c.client.endpoint(c.basePath, data.ClusterName, "failureDomains", data.DomainName)
-	return c.client.delete(endpoint)
+	endpoint := c.pulsar.endpoint(c.basePath, data.ClusterName, "failureDomains", data.DomainName)
+	return c.pulsar.Client.Delete(endpoint)
 }
 func (c *clusters) UpdateFailureDomain(data utils.FailureDomainData) error {
-	endpoint := c.client.endpoint(c.basePath, data.ClusterName, "failureDomains", data.DomainName)
-	return c.client.post(endpoint, &data)
+	endpoint := c.pulsar.endpoint(c.basePath, data.ClusterName, "failureDomains", data.DomainName)
+	return c.pulsar.Client.Post(endpoint, &data)
 }
diff --git a/pkg/pulsar/functions.go b/pkg/pulsar/functions.go
index 5cb79d289..41b82df69 100644
--- a/pkg/pulsar/functions.go
+++ b/pkg/pulsar/functions.go
@@ -113,14 +113,14 @@ type Functions interface {
 }
 
 type functions struct {
-	client   *client
+	pulsar   *pulsarClient
 	basePath string
 }
 
 // Functions is used to access the functions endpoints
-func (c *client) Functions() Functions {
+func (c *pulsarClient) Functions() Functions {
 	return &functions{
-		client:   c,
+		pulsar:   c,
 		basePath: "/functions",
 	}
 }
@@ -140,7 +140,7 @@ func (f *functions) createTextFromFiled(w *multipart.Writer, value string) (io.W
 }
 
 func (f *functions) CreateFunc(funcConf *utils.FunctionConfig, fileName string) error {
-	endpoint := f.client.endpoint(f.basePath, funcConf.Tenant, funcConf.Namespace, funcConf.Name)
+	endpoint := f.pulsar.endpoint(f.basePath, funcConf.Tenant, funcConf.Namespace, funcConf.Name)
 
 	// buffer to store our request as bytes
 	bodyBuf := bytes.NewBufferString("")
@@ -190,7 +190,7 @@ func (f *functions) CreateFunc(funcConf *utils.FunctionConfig, fileName string)
 	}
 
 	contentType := multiPartWriter.FormDataContentType()
-	err = f.client.postWithMultiPart(endpoint, nil, bodyBuf, contentType)
+	err = f.pulsar.Client.PostWithMultiPart(endpoint, nil, bodyBuf, contentType)
 	if err != nil {
 		return err
 	}
@@ -199,7 +199,7 @@ func (f *functions) CreateFunc(funcConf *utils.FunctionConfig, fileName string)
 }
 
 func (f *functions) CreateFuncWithURL(funcConf *utils.FunctionConfig, pkgURL string) error {
-	endpoint := f.client.endpoint(f.basePath, funcConf.Tenant, funcConf.Namespace, funcConf.Name)
+	endpoint := f.pulsar.endpoint(f.basePath, funcConf.Tenant, funcConf.Namespace, funcConf.Name)
 	// buffer to store our request as bytes
 	bodyBuf := bytes.NewBufferString("")
 
@@ -235,7 +235,7 @@ func (f *functions) CreateFuncWithURL(funcConf *utils.FunctionConfig, pkgURL str
 	}
 
 	contentType := multiPartWriter.FormDataContentType()
-	err = f.client.postWithMultiPart(endpoint, nil, bodyBuf, contentType)
+	err = f.pulsar.Client.PostWithMultiPart(endpoint, nil, bodyBuf, contentType)
 	if err != nil {
 		return err
 	}
@@ -244,63 +244,63 @@ func (f *functions) CreateFuncWithURL(funcConf *utils.FunctionConfig, pkgURL str
 }
 
 func (f *functions) StopFunction(tenant, namespace, name string) error {
-	endpoint := f.client.endpoint(f.basePath, tenant, namespace, name)
-	return f.client.post(endpoint+"/stop", "")
+	endpoint := f.pulsar.endpoint(f.basePath, tenant, namespace, name)
+	return f.pulsar.Client.Post(endpoint+"/stop", "")
 }
 
 func (f *functions) StopFunctionWithID(tenant, namespace, name string, instanceID int) error {
 	id := fmt.Sprintf("%d", instanceID)
-	endpoint := f.client.endpoint(f.basePath, tenant, namespace, name, id)
+	endpoint := f.pulsar.endpoint(f.basePath, tenant, namespace, name, id)
 
-	return f.client.post(endpoint+"/stop", "")
+	return f.pulsar.Client.Post(endpoint+"/stop", "")
 }
 
 func (f *functions) DeleteFunction(tenant, namespace, name string) error {
-	endpoint := f.client.endpoint(f.basePath, tenant, namespace, name)
-	return f.client.delete(endpoint)
+	endpoint := f.pulsar.endpoint(f.basePath, tenant, namespace, name)
+	return f.pulsar.Client.Delete(endpoint)
 }
 
 func (f *functions) StartFunction(tenant, namespace, name string) error {
-	endpoint := f.client.endpoint(f.basePath, tenant, namespace, name)
-	return f.client.post(endpoint+"/start", "")
+	endpoint := f.pulsar.endpoint(f.basePath, tenant, namespace, name)
+	return f.pulsar.Client.Post(endpoint+"/start", "")
 }
 
 func (f *functions) StartFunctionWithID(tenant, namespace, name string, instanceID int) error {
 	id := fmt.Sprintf("%d", instanceID)
-	endpoint := f.client.endpoint(f.basePath, tenant, namespace, name, id)
+	endpoint := f.pulsar.endpoint(f.basePath, tenant, namespace, name, id)
 
-	return f.client.post(endpoint+"/start", "")
+	return f.pulsar.Client.Post(endpoint+"/start", "")
 }
 
 func (f *functions) RestartFunction(tenant, namespace, name string) error {
-	endpoint := f.client.endpoint(f.basePath, tenant, namespace, name)
-	return f.client.post(endpoint+"/restart", "")
+	endpoint := f.pulsar.endpoint(f.basePath, tenant, namespace, name)
+	return f.pulsar.Client.Post(endpoint+"/restart", "")
 }
 
 func (f *functions) RestartFunctionWithID(tenant, namespace, name string, instanceID int) error {
 	id := fmt.Sprintf("%d", instanceID)
-	endpoint := f.client.endpoint(f.basePath, tenant, namespace, name, id)
+	endpoint := f.pulsar.endpoint(f.basePath, tenant, namespace, name, id)
 
-	return f.client.post(endpoint+"/restart", "")
+	return f.pulsar.Client.Post(endpoint+"/restart", "")
 }
 
 func (f *functions) GetFunctions(tenant, namespace string) ([]string, error) {
 	var functions []string
-	endpoint := f.client.endpoint(f.basePath, tenant, namespace)
-	err := f.client.get(endpoint, &functions)
+	endpoint := f.pulsar.endpoint(f.basePath, tenant, namespace)
+	err := f.pulsar.Client.Get(endpoint, &functions)
 	return functions, err
 }
 
 func (f *functions) GetFunction(tenant, namespace, name string) (utils.FunctionConfig, error) {
 	var functionConfig utils.FunctionConfig
-	endpoint := f.client.endpoint(f.basePath, tenant, namespace, name)
-	err := f.client.get(endpoint, &functionConfig)
+	endpoint := f.pulsar.endpoint(f.basePath, tenant, namespace, name)
+	err := f.pulsar.Client.Get(endpoint, &functionConfig)
 	return functionConfig, err
 }
 
 func (f *functions) UpdateFunction(functionConfig *utils.FunctionConfig, fileName string,
 	updateOptions *utils.UpdateOptions) error {
-	endpoint := f.client.endpoint(f.basePath, functionConfig.Tenant, functionConfig.Namespace, functionConfig.Name)
+	endpoint := f.pulsar.endpoint(f.basePath, functionConfig.Tenant, functionConfig.Namespace, functionConfig.Name)
 	// buffer to store our request as bytes
 	bodyBuf := bytes.NewBufferString("")
 
@@ -366,7 +366,7 @@ func (f *functions) UpdateFunction(functionConfig *utils.FunctionConfig, fileNam
 	}
 
 	contentType := multiPartWriter.FormDataContentType()
-	err = f.client.putWithMultiPart(endpoint, bodyBuf, contentType)
+	err = f.pulsar.Client.PutWithMultiPart(endpoint, bodyBuf, contentType)
 	if err != nil {
 		return err
 	}
@@ -376,7 +376,7 @@ func (f *functions) UpdateFunction(functionConfig *utils.FunctionConfig, fileNam
 
 func (f *functions) UpdateFunctionWithURL(functionConfig *utils.FunctionConfig, pkgURL string,
 	updateOptions *utils.UpdateOptions) error {
-	endpoint := f.client.endpoint(f.basePath, functionConfig.Tenant, functionConfig.Namespace, functionConfig.Name)
+	endpoint := f.pulsar.endpoint(f.basePath, functionConfig.Tenant, functionConfig.Namespace, functionConfig.Name)
 	// buffer to store our request as bytes
 	bodyBuf := bytes.NewBufferString("")
 
@@ -431,7 +431,7 @@ func (f *functions) UpdateFunctionWithURL(functionConfig *utils.FunctionConfig,
 	}
 
 	contentType := multiPartWriter.FormDataContentType()
-	err = f.client.putWithMultiPart(endpoint, bodyBuf, contentType)
+	err = f.pulsar.Client.PutWithMultiPart(endpoint, bodyBuf, contentType)
 	if err != nil {
 		return err
 	}
@@ -441,8 +441,8 @@ func (f *functions) UpdateFunctionWithURL(functionConfig *utils.FunctionConfig,
 
 func (f *functions) GetFunctionStatus(tenant, namespace, name string) (utils.FunctionStatus, error) {
 	var functionStatus utils.FunctionStatus
-	endpoint := f.client.endpoint(f.basePath, tenant, namespace, name)
-	err := f.client.get(endpoint+"/status", &functionStatus)
+	endpoint := f.pulsar.endpoint(f.basePath, tenant, namespace, name)
+	err := f.pulsar.Client.Get(endpoint+"/status", &functionStatus)
 	return functionStatus, err
 }
 
@@ -450,15 +450,15 @@ func (f *functions) GetFunctionStatusWithInstanceID(tenant, namespace, name stri
 	instanceID int) (utils.FunctionInstanceStatusData, error) {
 	var functionInstanceStatusData utils.FunctionInstanceStatusData
 	id := fmt.Sprintf("%d", instanceID)
-	endpoint := f.client.endpoint(f.basePath, tenant, namespace, name, id)
-	err := f.client.get(endpoint+"/status", &functionInstanceStatusData)
+	endpoint := f.pulsar.endpoint(f.basePath, tenant, namespace, name, id)
+	err := f.pulsar.Client.Get(endpoint+"/status", &functionInstanceStatusData)
 	return functionInstanceStatusData, err
 }
 
 func (f *functions) GetFunctionStats(tenant, namespace, name string) (utils.FunctionStats, error) {
 	var functionStats utils.FunctionStats
-	endpoint := f.client.endpoint(f.basePath, tenant, namespace, name)
-	err := f.client.get(endpoint+"/stats", &functionStats)
+	endpoint := f.pulsar.endpoint(f.basePath, tenant, namespace, name)
+	err := f.pulsar.Client.Get(endpoint+"/stats", &functionStats)
 	return functionStats, err
 }
 
@@ -466,20 +466,20 @@ func (f *functions) GetFunctionStatsWithInstanceID(tenant, namespace, name strin
 	instanceID int) (utils.FunctionInstanceStatsData, error) {
 	var functionInstanceStatsData utils.FunctionInstanceStatsData
 	id := fmt.Sprintf("%d", instanceID)
-	endpoint := f.client.endpoint(f.basePath, tenant, namespace, name, id)
-	err := f.client.get(endpoint+"/stats", &functionInstanceStatsData)
+	endpoint := f.pulsar.endpoint(f.basePath, tenant, namespace, name, id)
+	err := f.pulsar.Client.Get(endpoint+"/stats", &functionInstanceStatsData)
 	return functionInstanceStatsData, err
 }
 
 func (f *functions) GetFunctionState(tenant, namespace, name, key string) (utils.FunctionState, error) {
 	var functionState utils.FunctionState
-	endpoint := f.client.endpoint(f.basePath, tenant, namespace, name, "state", key)
-	err := f.client.get(endpoint, &functionState)
+	endpoint := f.pulsar.endpoint(f.basePath, tenant, namespace, name, "state", key)
+	err := f.pulsar.Client.Get(endpoint, &functionState)
 	return functionState, err
 }
 
 func (f *functions) PutFunctionState(tenant, namespace, name string, state utils.FunctionState) error {
-	endpoint := f.client.endpoint(f.basePath, tenant, namespace, name, "state", state.Key)
+	endpoint := f.pulsar.endpoint(f.basePath, tenant, namespace, name, "state", state.Key)
 
 	// buffer to store our request as bytes
 	bodyBuf := bytes.NewBufferString("")
@@ -511,7 +511,7 @@ func (f *functions) PutFunctionState(tenant, namespace, name string, state utils
 
 	contentType := multiPartWriter.FormDataContentType()
 
-	err = f.client.postWithMultiPart(endpoint, nil, bodyBuf, contentType)
+	err = f.pulsar.Client.PostWithMultiPart(endpoint, nil, bodyBuf, contentType)
 
 	if err != nil {
 		return err
@@ -521,7 +521,7 @@ func (f *functions) PutFunctionState(tenant, namespace, name string, state utils
 }
 
 func (f *functions) TriggerFunction(tenant, namespace, name, topic, triggerValue, triggerFile string) (string, error) {
-	endpoint := f.client.endpoint(f.basePath, tenant, namespace, name, "trigger")
+	endpoint := f.pulsar.endpoint(f.basePath, tenant, namespace, name, "trigger")
 
 	// buffer to store our request as bytes
 	bodyBuf := bytes.NewBufferString("")
@@ -580,7 +580,7 @@ func (f *functions) TriggerFunction(tenant, namespace, name, topic, triggerValue
 
 	contentType := multiPartWriter.FormDataContentType()
 	var str string
-	err := f.client.postWithMultiPart(endpoint, &str, bodyBuf, contentType)
+	err := f.pulsar.Client.PostWithMultiPart(endpoint, &str, bodyBuf, contentType)
 	if err != nil {
 		return "", err
 	}
diff --git a/pkg/pulsar/functions_worker.go b/pkg/pulsar/functions_worker.go
index d7ea5d3d1..dfc52427e 100644
--- a/pkg/pulsar/functions_worker.go
+++ b/pkg/pulsar/functions_worker.go
@@ -39,23 +39,23 @@ type FunctionsWorker interface {
 }
 
 type worker struct {
-	client          *client
+	pulsar          *pulsarClient
 	workerPath      string
 	workerStatsPath string
 }
 
-func (c *client) FunctionsWorker() FunctionsWorker {
+func (c *pulsarClient) FunctionsWorker() FunctionsWorker {
 	return &worker{
-		client:          c,
+		pulsar:          c,
 		workerPath:      "/worker",
 		workerStatsPath: "/worker-stats",
 	}
 }
 
 func (w *worker) GetFunctionsStats() ([]*utils.WorkerFunctionInstanceStats, error) {
-	endpoint := w.client.endpoint(w.workerStatsPath, "functionsmetrics")
+	endpoint := w.pulsar.endpoint(w.workerStatsPath, "functionsmetrics")
 	var workerStats []*utils.WorkerFunctionInstanceStats
-	err := w.client.get(endpoint, &workerStats)
+	err := w.pulsar.Client.Get(endpoint, &workerStats)
 	if err != nil {
 		return nil, err
 	}
@@ -63,9 +63,9 @@ func (w *worker) GetFunctionsStats() ([]*utils.WorkerFunctionInstanceStats, erro
 }
 
 func (w *worker) GetMetrics() ([]*utils.Metrics, error) {
-	endpoint := w.client.endpoint(w.workerStatsPath, "metrics")
+	endpoint := w.pulsar.endpoint(w.workerStatsPath, "metrics")
 	var metrics []*utils.Metrics
-	err := w.client.get(endpoint, &metrics)
+	err := w.pulsar.Client.Get(endpoint, &metrics)
 	if err != nil {
 		return nil, err
 	}
@@ -73,9 +73,9 @@ func (w *worker) GetMetrics() ([]*utils.Metrics, error) {
 }
 
 func (w *worker) GetCluster() ([]*utils.WorkerInfo, error) {
-	endpoint := w.client.endpoint(w.workerPath, "cluster")
+	endpoint := w.pulsar.endpoint(w.workerPath, "cluster")
 	var workersInfo []*utils.WorkerInfo
-	err := w.client.get(endpoint, &workersInfo)
+	err := w.pulsar.Client.Get(endpoint, &workersInfo)
 	if err != nil {
 		return nil, err
 	}
@@ -83,9 +83,9 @@ func (w *worker) GetCluster() ([]*utils.WorkerInfo, error) {
 }
 
 func (w *worker) GetClusterLeader() (*utils.WorkerInfo, error) {
-	endpoint := w.client.endpoint(w.workerPath, "cluster", "leader")
+	endpoint := w.pulsar.endpoint(w.workerPath, "cluster", "leader")
 	var workerInfo utils.WorkerInfo
-	err := w.client.get(endpoint, &workerInfo)
+	err := w.pulsar.Client.Get(endpoint, &workerInfo)
 	if err != nil {
 		return nil, err
 	}
@@ -93,9 +93,9 @@ func (w *worker) GetClusterLeader() (*utils.WorkerInfo, error) {
 }
 
 func (w *worker) GetAssignments() (map[string][]string, error) {
-	endpoint := w.client.endpoint(w.workerPath, "assignments")
+	endpoint := w.pulsar.endpoint(w.workerPath, "assignments")
 	var assignments map[string][]string
-	err := w.client.get(endpoint, &assignments)
+	err := w.pulsar.Client.Get(endpoint, &assignments)
 	if err != nil {
 		return nil, err
 	}
diff --git a/pkg/pulsar/namespace.go b/pkg/pulsar/namespace.go
index 7f7e801d5..86fa1fe6b 100644
--- a/pkg/pulsar/namespace.go
+++ b/pkg/pulsar/namespace.go
@@ -250,22 +250,22 @@ type Namespaces interface {
 }
 
 type namespaces struct {
-	client   *client
+	pulsar   *pulsarClient
 	basePath string
 }
 
 // Namespaces is used to access the namespaces endpoints
-func (c *client) Namespaces() Namespaces {
+func (c *pulsarClient) Namespaces() Namespaces {
 	return &namespaces{
-		client:   c,
+		pulsar:   c,
 		basePath: "/namespaces",
 	}
 }
 
 func (n *namespaces) GetNamespaces(tenant string) ([]string, error) {
 	var namespaces []string
-	endpoint := n.client.endpoint(n.basePath, tenant)
-	err := n.client.get(endpoint, &namespaces)
+	endpoint := n.pulsar.endpoint(n.basePath, tenant)
+	err := n.pulsar.Client.Get(endpoint, &namespaces)
 	return namespaces, err
 }
 
@@ -275,8 +275,8 @@ func (n *namespaces) GetTopics(namespace string) ([]string, error) {
 	if err != nil {
 		return nil, err
 	}
-	endpoint := n.client.endpoint(n.basePath, ns.String(), "topics")
-	err = n.client.get(endpoint, &topics)
+	endpoint := n.pulsar.endpoint(n.basePath, ns.String(), "topics")
+	err = n.pulsar.Client.Get(endpoint, &topics)
 	return topics, err
 }
 
@@ -286,8 +286,8 @@ func (n *namespaces) GetPolicies(namespace string) (*utils.Policies, error) {
 	if err != nil {
 		return nil, err
 	}
-	endpoint := n.client.endpoint(n.basePath, ns.String())
-	err = n.client.get(endpoint, &police)
+	endpoint := n.pulsar.endpoint(n.basePath, ns.String())
+	err = n.pulsar.Client.Get(endpoint, &police)
 	return &police, err
 }
 
@@ -300,8 +300,8 @@ func (n *namespaces) CreateNsWithPolices(namespace string, policies utils.Polici
 	if err != nil {
 		return err
 	}
-	endpoint := n.client.endpoint(n.basePath, ns.String())
-	return n.client.put(endpoint, &policies)
+	endpoint := n.pulsar.endpoint(n.basePath, ns.String())
+	return n.pulsar.Client.Put(endpoint, &policies)
 }
 
 func (n *namespaces) CreateNsWithBundlesData(namespace string, bundleData *utils.BundlesData) error {
@@ -309,11 +309,11 @@ func (n *namespaces) CreateNsWithBundlesData(namespace string, bundleData *utils
 	if err != nil {
 		return err
 	}
-	endpoint := n.client.endpoint(n.basePath, ns.String())
+	endpoint := n.pulsar.endpoint(n.basePath, ns.String())
 	polices := new(utils.Policies)
 	polices.Bundles = bundleData
 
-	return n.client.put(endpoint, &polices)
+	return n.pulsar.Client.Put(endpoint, &polices)
 }
 
 func (n *namespaces) CreateNamespace(namespace string) error {
@@ -321,8 +321,8 @@ func (n *namespaces) CreateNamespace(namespace string) error {
 	if err != nil {
 		return err
 	}
-	endpoint := n.client.endpoint(n.basePath, ns.String())
-	return n.client.put(endpoint, nil)
+	endpoint := n.pulsar.endpoint(n.basePath, ns.String())
+	return n.pulsar.Client.Put(endpoint, nil)
 }
 
 func (n *namespaces) DeleteNamespace(namespace string) error {
@@ -330,8 +330,8 @@ func (n *namespaces) DeleteNamespace(namespace string) error {
 	if err != nil {
 		return err
 	}
-	endpoint := n.client.endpoint(n.basePath, ns.String())
-	return n.client.delete(endpoint)
+	endpoint := n.pulsar.endpoint(n.basePath, ns.String())
+	return n.pulsar.Client.Delete(endpoint)
 }
 
 func (n *namespaces) DeleteNamespaceBundle(namespace string, bundleRange string) error {
@@ -339,8 +339,8 @@ func (n *namespaces) DeleteNamespaceBundle(namespace string, bundleRange string)
 	if err != nil {
 		return err
 	}
-	endpoint := n.client.endpoint(n.basePath, ns.String(), bundleRange)
-	return n.client.delete(endpoint)
+	endpoint := n.pulsar.endpoint(n.basePath, ns.String(), bundleRange)
+	return n.pulsar.Client.Delete(endpoint)
 }
 
 func (n *namespaces) GetNamespaceMessageTTL(namespace string) (int, error) {
@@ -349,8 +349,8 @@ func (n *namespaces) GetNamespaceMessageTTL(namespace string) (int, error) {
 	if err != nil {
 		return 0, err
 	}
-	endpoint := n.client.endpoint(n.basePath, nsName.String(), "messageTTL")
-	err = n.client.get(endpoint, &ttl)
+	endpoint := n.pulsar.endpoint(n.basePath, nsName.String(), "messageTTL")
+	err = n.pulsar.Client.Get(endpoint, &ttl)
 	return ttl, err
 }
 
@@ -360,8 +360,8 @@ func (n *namespaces) SetNamespaceMessageTTL(namespace string, ttlInSeconds int)
 		return err
 	}
 
-	endpoint := n.client.endpoint(n.basePath, nsName.String(), "messageTTL")
-	return n.client.post(endpoint, &ttlInSeconds)
+	endpoint := n.pulsar.endpoint(n.basePath, nsName.String(), "messageTTL")
+	return n.pulsar.Client.Post(endpoint, &ttlInSeconds)
 }
 
 func (n *namespaces) SetRetention(namespace string, policy utils.RetentionPolicies) error {
@@ -369,8 +369,8 @@ func (n *namespaces) SetRetention(namespace string, policy utils.RetentionPolici
 	if err != nil {
 		return err
 	}
-	endpoint := n.client.endpoint(n.basePath, nsName.String(), "retention")
-	return n.client.post(endpoint, &policy)
+	endpoint := n.pulsar.endpoint(n.basePath, nsName.String(), "retention")
+	return n.pulsar.Client.Post(endpoint, &policy)
 }
 
 func (n *namespaces) GetRetention(namespace string) (*utils.RetentionPolicies, error) {
@@ -379,8 +379,8 @@ func (n *namespaces) GetRetention(namespace string) (*utils.RetentionPolicies, e
 	if err != nil {
 		return nil, err
 	}
-	endpoint := n.client.endpoint(n.basePath, nsName.String(), "retention")
-	err = n.client.get(endpoint, &policy)
+	endpoint := n.pulsar.endpoint(n.basePath, nsName.String(), "retention")
+	err = n.pulsar.Client.Get(endpoint, &policy)
 	return &policy, err
 }
 
@@ -390,8 +390,8 @@ func (n *namespaces) GetBacklogQuotaMap(namespace string) (map[utils.BacklogQuot
 	if err != nil {
 		return nil, err
 	}
-	endpoint := n.client.endpoint(n.basePath, nsName.String(), "backlogQuotaMap")
-	err = n.client.get(endpoint, &backlogQuotaMap)
+	endpoint := n.pulsar.endpoint(n.basePath, nsName.String(), "backlogQuotaMap")
+	err = n.pulsar.Client.Get(endpoint, &backlogQuotaMap)
 	return backlogQuotaMap, err
 }
 
@@ -400,8 +400,8 @@ func (n *namespaces) SetBacklogQuota(namespace string, backlogQuota utils.Backlo
 	if err != nil {
 		return err
 	}
-	endpoint := n.client.endpoint(n.basePath, nsName.String(), "backlogQuota")
-	return n.client.post(endpoint, &backlogQuota)
+	endpoint := n.pulsar.endpoint(n.basePath, nsName.String(), "backlogQuota")
+	return n.pulsar.Client.Post(endpoint, &backlogQuota)
 }
 
 func (n *namespaces) RemoveBacklogQuota(namespace string) error {
@@ -409,21 +409,21 @@ func (n *namespaces) RemoveBacklogQuota(namespace string) error {
 	if err != nil {
 		return err
 	}
-	endpoint := n.client.endpoint(n.basePath, nsName.String(), "backlogQuota")
+	endpoint := n.pulsar.endpoint(n.basePath, nsName.String(), "backlogQuota")
 	params := map[string]string{
 		"backlogQuotaType": string(utils.DestinationStorage),
 	}
-	return n.client.deleteWithQueryParams(endpoint, nil, params)
+	return n.pulsar.Client.DeleteWithQueryParams(endpoint, params)
 }
 
 func (n *namespaces) SetSchemaValidationEnforced(namespace utils.NameSpaceName, schemaValidationEnforced bool) error {
-	endpoint := n.client.endpoint(n.basePath, namespace.String(), "schemaValidationEnforced")
-	return n.client.post(endpoint, schemaValidationEnforced)
+	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "schemaValidationEnforced")
+	return n.pulsar.Client.Post(endpoint, schemaValidationEnforced)
 }
 
 func (n *namespaces) GetSchemaValidationEnforced(namespace utils.NameSpaceName) (bool, error) {
-	endpoint := n.client.endpoint(n.basePath, namespace.String(), "schemaValidationEnforced")
-	r, err := n.client.getWithQueryParams(endpoint, nil, nil, false)
+	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "schemaValidationEnforced")
+	r, err := n.pulsar.Client.GetWithQueryParams(endpoint, nil, nil, false)
 	if err != nil {
 		return false, err
 	}
@@ -432,15 +432,15 @@ func (n *namespaces) GetSchemaValidationEnforced(namespace utils.NameSpaceName)
 
 func (n *namespaces) SetSchemaAutoUpdateCompatibilityStrategy(namespace utils.NameSpaceName,
 	strategy utils.SchemaCompatibilityStrategy) error {
-	endpoint := n.client.endpoint(n.basePath, namespace.String(), "schemaAutoUpdateCompatibilityStrategy")
-	return n.client.put(endpoint, strategy.String())
+	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "schemaAutoUpdateCompatibilityStrategy")
+	return n.pulsar.Client.Put(endpoint, strategy.String())
 }
 
 func (n *namespaces) GetSchemaAutoUpdateCompatibilityStrategy(namespace utils.NameSpaceName) (
 	utils.SchemaCompatibilityStrategy, error) {
 
-	endpoint := n.client.endpoint(n.basePath, namespace.String(), "schemaAutoUpdateCompatibilityStrategy")
-	b, err := n.client.getWithQueryParams(endpoint, nil, nil, false)
+	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "schemaAutoUpdateCompatibilityStrategy")
+	b, err := n.pulsar.Client.GetWithQueryParams(endpoint, nil, nil, false)
 	if err != nil {
 		return "", err
 	}
@@ -452,18 +452,18 @@ func (n *namespaces) GetSchemaAutoUpdateCompatibilityStrategy(namespace utils.Na
 }
 
 func (n *namespaces) ClearOffloadDeleteLag(namespace utils.NameSpaceName) error {
-	endpoint := n.client.endpoint(n.basePath, namespace.String(), "offloadDeletionLagMs")
-	return n.client.delete(endpoint)
+	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "offloadDeletionLagMs")
+	return n.pulsar.Client.Delete(endpoint)
 }
 
 func (n *namespaces) SetOffloadDeleteLag(namespace utils.NameSpaceName, timeMs int64) error {
-	endpoint := n.client.endpoint(n.basePath, namespace.String(), "offloadDeletionLagMs")
-	return n.client.put(endpoint, timeMs)
+	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "offloadDeletionLagMs")
+	return n.pulsar.Client.Put(endpoint, timeMs)
 }
 
 func (n *namespaces) GetOffloadDeleteLag(namespace utils.NameSpaceName) (int64, error) {
-	endpoint := n.client.endpoint(n.basePath, namespace.String(), "offloadDeletionLagMs")
-	b, err := n.client.getWithQueryParams(endpoint, nil, nil, false)
+	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "offloadDeletionLagMs")
+	b, err := n.pulsar.Client.GetWithQueryParams(endpoint, nil, nil, false)
 	if err != nil {
 		return -1, err
 	}
@@ -471,13 +471,13 @@ func (n *namespaces) GetOffloadDeleteLag(namespace utils.NameSpaceName) (int64,
 }
 
 func (n *namespaces) SetMaxConsumersPerSubscription(namespace utils.NameSpaceName, max int) error {
-	endpoint := n.client.endpoint(n.basePath, namespace.String(), "maxConsumersPerSubscription")
-	return n.client.post(endpoint, max)
+	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "maxConsumersPerSubscription")
+	return n.pulsar.Client.Post(endpoint, max)
 }
 
 func (n *namespaces) GetMaxConsumersPerSubscription(namespace utils.NameSpaceName) (int, error) {
-	endpoint := n.client.endpoint(n.basePath, namespace.String(), "maxConsumersPerSubscription")
-	b, err := n.client.getWithQueryParams(endpoint, nil, nil, false)
+	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "maxConsumersPerSubscription")
+	b, err := n.pulsar.Client.GetWithQueryParams(endpoint, nil, nil, false)
 	if err != nil {
 		return -1, err
 	}
@@ -485,13 +485,13 @@ func (n *namespaces) GetMaxConsumersPerSubscription(namespace utils.NameSpaceNam
 }
 
 func (n *namespaces) SetOffloadThreshold(namespace utils.NameSpaceName, threshold int64) error {
-	endpoint := n.client.endpoint(n.basePath, namespace.String(), "offloadThreshold")
-	return n.client.put(endpoint, threshold)
+	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "offloadThreshold")
+	return n.pulsar.Client.Put(endpoint, threshold)
 }
 
 func (n *namespaces) GetOffloadThreshold(namespace utils.NameSpaceName) (int64, error) {
-	endpoint := n.client.endpoint(n.basePath, namespace.String(), "offloadThreshold")
-	b, err := n.client.getWithQueryParams(endpoint, nil, nil, false)
+	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "offloadThreshold")
+	b, err := n.pulsar.Client.GetWithQueryParams(endpoint, nil, nil, false)
 	if err != nil {
 		return -1, err
 	}
@@ -499,13 +499,13 @@ func (n *namespaces) GetOffloadThreshold(namespace utils.NameSpaceName) (int64,
 }
 
 func (n *namespaces) SetMaxConsumersPerTopic(namespace utils.NameSpaceName, max int) error {
-	endpoint := n.client.endpoint(n.basePath, namespace.String(), "maxConsumersPerTopic")
-	return n.client.post(endpoint, max)
+	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "maxConsumersPerTopic")
+	return n.pulsar.Client.Post(endpoint, max)
 }
 
 func (n *namespaces) GetMaxConsumersPerTopic(namespace utils.NameSpaceName) (int, error) {
-	endpoint := n.client.endpoint(n.basePath, namespace.String(), "maxConsumersPerTopic")
-	b, err := n.client.getWithQueryParams(endpoint, nil, nil, false)
+	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "maxConsumersPerTopic")
+	b, err := n.pulsar.Client.GetWithQueryParams(endpoint, nil, nil, false)
 	if err != nil {
 		return -1, err
 	}
@@ -513,13 +513,13 @@ func (n *namespaces) GetMaxConsumersPerTopic(namespace utils.NameSpaceName) (int
 }
 
 func (n *namespaces) SetCompactionThreshold(namespace utils.NameSpaceName, threshold int64) error {
-	endpoint := n.client.endpoint(n.basePath, namespace.String(), "compactionThreshold")
-	return n.client.put(endpoint, threshold)
+	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "compactionThreshold")
+	return n.pulsar.Client.Put(endpoint, threshold)
 }
 
 func (n *namespaces) GetCompactionThreshold(namespace utils.NameSpaceName) (int64, error) {
-	endpoint := n.client.endpoint(n.basePath, namespace.String(), "compactionThreshold")
-	b, err := n.client.getWithQueryParams(endpoint, nil, nil, false)
+	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "compactionThreshold")
+	b, err := n.pulsar.Client.GetWithQueryParams(endpoint, nil, nil, false)
 	if err != nil {
 		return -1, err
 	}
@@ -527,13 +527,13 @@ func (n *namespaces) GetCompactionThreshold(namespace utils.NameSpaceName) (int6
 }
 
 func (n *namespaces) SetMaxProducersPerTopic(namespace utils.NameSpaceName, max int) error {
-	endpoint := n.client.endpoint(n.basePath, namespace.String(), "maxProducersPerTopic")
-	return n.client.post(endpoint, max)
+	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "maxProducersPerTopic")
+	return n.pulsar.Client.Post(endpoint, max)
 }
 
 func (n *namespaces) GetMaxProducersPerTopic(namespace utils.NameSpaceName) (int, error) {
-	endpoint := n.client.endpoint(n.basePath, namespace.String(), "maxProducersPerTopic")
-	b, err := n.client.getWithQueryParams(endpoint, nil, nil, false)
+	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "maxProducersPerTopic")
+	b, err := n.pulsar.Client.GetWithQueryParams(endpoint, nil, nil, false)
 	if err != nil {
 		return -1, err
 	}
@@ -546,8 +546,8 @@ func (n *namespaces) GetNamespaceReplicationClusters(namespace string) ([]string
 	if err != nil {
 		return nil, err
 	}
-	endpoint := n.client.endpoint(n.basePath, nsName.String(), "replication")
-	err = n.client.get(endpoint, &data)
+	endpoint := n.pulsar.endpoint(n.basePath, nsName.String(), "replication")
+	err = n.pulsar.Client.Get(endpoint, &data)
 	return data, err
 }
 
@@ -556,8 +556,8 @@ func (n *namespaces) SetNamespaceReplicationClusters(namespace string, clusterId
 	if err != nil {
 		return err
 	}
-	endpoint := n.client.endpoint(n.basePath, nsName.String(), "replication")
-	return n.client.post(endpoint, &clusterIds)
+	endpoint := n.pulsar.endpoint(n.basePath, nsName.String(), "replication")
+	return n.pulsar.Client.Post(endpoint, &clusterIds)
 }
 
 func (n *namespaces) SetNamespaceAntiAffinityGroup(namespace string, namespaceAntiAffinityGroup string) error {
@@ -565,17 +565,17 @@ func (n *namespaces) SetNamespaceAntiAffinityGroup(namespace string, namespaceAn
 	if err != nil {
 		return err
 	}
-	endpoint := n.client.endpoint(n.basePath, nsName.String(), "antiAffinity")
-	return n.client.post(endpoint, namespaceAntiAffinityGroup)
+	endpoint := n.pulsar.endpoint(n.basePath, nsName.String(), "antiAffinity")
+	return n.pulsar.Client.Post(endpoint, namespaceAntiAffinityGroup)
 }
 
 func (n *namespaces) GetAntiAffinityNamespaces(tenant, cluster, namespaceAntiAffinityGroup string) ([]string, error) {
 	var data []string
-	endpoint := n.client.endpoint(n.basePath, cluster, "antiAffinity", namespaceAntiAffinityGroup)
+	endpoint := n.pulsar.endpoint(n.basePath, cluster, "antiAffinity", namespaceAntiAffinityGroup)
 	params := map[string]string{
 		"property": tenant,
 	}
-	_, err := n.client.getWithQueryParams(endpoint, &data, params, false)
+	_, err := n.pulsar.Client.GetWithQueryParams(endpoint, &data, params, false)
 	return data, err
 }
 
@@ -584,8 +584,8 @@ func (n *namespaces) GetNamespaceAntiAffinityGroup(namespace string) (string, er
 	if err != nil {
 		return "", err
 	}
-	endpoint := n.client.endpoint(n.basePath, nsName.String(), "antiAffinity")
-	data, err := n.client.getWithQueryParams(endpoint, nil, nil, false)
+	endpoint := n.pulsar.endpoint(n.basePath, nsName.String(), "antiAffinity")
+	data, err := n.pulsar.Client.GetWithQueryParams(endpoint, nil, nil, false)
 	return string(data), err
 }
 
@@ -594,8 +594,8 @@ func (n *namespaces) DeleteNamespaceAntiAffinityGroup(namespace string) error {
 	if err != nil {
 		return err
 	}
-	endpoint := n.client.endpoint(n.basePath, nsName.String(), "antiAffinity")
-	return n.client.delete(endpoint)
+	endpoint := n.pulsar.endpoint(n.basePath, nsName.String(), "antiAffinity")
+	return n.pulsar.Client.Delete(endpoint)
 }
 
 func (n *namespaces) SetDeduplicationStatus(namespace string, enableDeduplication bool) error {
@@ -603,8 +603,8 @@ func (n *namespaces) SetDeduplicationStatus(namespace string, enableDeduplicatio
 	if err != nil {
 		return err
 	}
-	endpoint := n.client.endpoint(n.basePath, nsName.String(), "deduplication")
-	return n.client.post(endpoint, enableDeduplication)
+	endpoint := n.pulsar.endpoint(n.basePath, nsName.String(), "deduplication")
+	return n.pulsar.Client.Post(endpoint, enableDeduplication)
 }
 
 func (n *namespaces) SetPersistence(namespace string, persistence utils.PersistencePolicies) error {
@@ -612,8 +612,8 @@ func (n *namespaces) SetPersistence(namespace string, persistence utils.Persiste
 	if err != nil {
 		return err
 	}
-	endpoint := n.client.endpoint(n.basePath, nsName.String(), "persistence")
-	return n.client.post(endpoint, &persistence)
+	endpoint := n.pulsar.endpoint(n.basePath, nsName.String(), "persistence")
+	return n.pulsar.Client.Post(endpoint, &persistence)
 }
 
 func (n *namespaces) SetBookieAffinityGroup(namespace string, bookieAffinityGroup utils.BookieAffinityGroupData) error {
@@ -621,8 +621,8 @@ func (n *namespaces) SetBookieAffinityGroup(namespace string, bookieAffinityGrou
 	if err != nil {
 		return err
 	}
-	endpoint := n.client.endpoint(n.basePath, nsName.String(), "persistence", "bookieAffinity")
-	return n.client.post(endpoint, &bookieAffinityGroup)
+	endpoint := n.pulsar.endpoint(n.basePath, nsName.String(), "persistence", "bookieAffinity")
+	return n.pulsar.Client.Post(endpoint, &bookieAffinityGroup)
 }
 
 func (n *namespaces) DeleteBookieAffinityGroup(namespace string) error {
@@ -630,8 +630,8 @@ func (n *namespaces) DeleteBookieAffinityGroup(namespace string) error {
 	if err != nil {
 		return err
 	}
-	endpoint := n.client.endpoint(n.basePath, nsName.String(), "persistence", "bookieAffinity")
-	return n.client.delete(endpoint)
+	endpoint := n.pulsar.endpoint(n.basePath, nsName.String(), "persistence", "bookieAffinity")
+	return n.pulsar.Client.Delete(endpoint)
 }
 
 func (n *namespaces) GetBookieAffinityGroup(namespace string) (*utils.BookieAffinityGroupData, error) {
@@ -640,8 +640,8 @@ func (n *namespaces) GetBookieAffinityGroup(namespace string) (*utils.BookieAffi
 	if err != nil {
 		return nil, err
 	}
-	endpoint := n.client.endpoint(n.basePath, nsName.String(), "persistence", "bookieAffinity")
-	err = n.client.get(endpoint, &data)
+	endpoint := n.pulsar.endpoint(n.basePath, nsName.String(), "persistence", "bookieAffinity")
+	err = n.pulsar.Client.Get(endpoint, &data)
 	return &data, err
 }
 
@@ -651,8 +651,8 @@ func (n *namespaces) GetPersistence(namespace string) (*utils.PersistencePolicie
 	if err != nil {
 		return nil, err
 	}
-	endpoint := n.client.endpoint(n.basePath, nsName.String(), "persistence")
-	err = n.client.get(endpoint, &persistence)
+	endpoint := n.pulsar.endpoint(n.basePath, nsName.String(), "persistence")
+	err = n.pulsar.Client.Get(endpoint, &persistence)
 	return &persistence, err
 }
 
@@ -661,8 +661,8 @@ func (n *namespaces) Unload(namespace string) error {
 	if err != nil {
 		return err
 	}
-	endpoint := n.client.endpoint(n.basePath, nsName.String(), "unload")
-	return n.client.put(endpoint, "")
+	endpoint := n.pulsar.endpoint(n.basePath, nsName.String(), "unload")
+	return n.pulsar.Client.Put(endpoint, "")
 }
 
 func (n *namespaces) UnloadNamespaceBundle(namespace, bundle string) error {
@@ -670,8 +670,8 @@ func (n *namespaces) UnloadNamespaceBundle(namespace, bundle string) error {
 	if err != nil {
 		return err
 	}
-	endpoint := n.client.endpoint(n.basePath, nsName.String(), bundle, "unload")
-	return n.client.put(endpoint, "")
+	endpoint := n.pulsar.endpoint(n.basePath, nsName.String(), bundle, "unload")
+	return n.pulsar.Client.Put(endpoint, "")
 }
 
 func (n *namespaces) SplitNamespaceBundle(namespace, bundle string, unloadSplitBundles bool) error {
@@ -679,132 +679,132 @@ func (n *namespaces) SplitNamespaceBundle(namespace, bundle string, unloadSplitB
 	if err != nil {
 		return err
 	}
-	endpoint := n.client.endpoint(n.basePath, nsName.String(), bundle, "split")
+	endpoint := n.pulsar.endpoint(n.basePath, nsName.String(), bundle, "split")
 	params := map[string]string{
 		"unload": strconv.FormatBool(unloadSplitBundles),
 	}
-	return n.client.putWithQueryParams(endpoint, "", nil, params)
+	return n.pulsar.Client.PutWithQueryParams(endpoint, "", nil, params)
 }
 
 func (n *namespaces) GetNamespacePermissions(namespace utils.NameSpaceName) (map[string][]common.AuthAction, error) {
-	endpoint := n.client.endpoint(n.basePath, namespace.String(), "permissions")
+	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "permissions")
 	var permissions map[string][]common.AuthAction
-	err := n.client.get(endpoint, &permissions)
+	err := n.pulsar.Client.Get(endpoint, &permissions)
 	return permissions, err
 }
 
 func (n *namespaces) GrantNamespacePermission(namespace utils.NameSpaceName, role string,
 	action []common.AuthAction) error {
-	endpoint := n.client.endpoint(n.basePath, namespace.String(), "permissions", role)
+	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "permissions", role)
 	s := make([]string, 0)
 	for _, v := range action {
 		s = append(s, v.String())
 	}
-	return n.client.post(endpoint, s)
+	return n.pulsar.Client.Post(endpoint, s)
 }
 
 func (n *namespaces) RevokeNamespacePermission(namespace utils.NameSpaceName, role string) error {
-	endpoint := n.client.endpoint(n.basePath, namespace.String(), "permissions", role)
-	return n.client.delete(endpoint)
+	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "permissions", role)
+	return n.pulsar.Client.Delete(endpoint)
 }
 
 func (n *namespaces) GrantSubPermission(namespace utils.NameSpaceName, sName string, roles []string) error {
-	endpoint := n.client.endpoint(n.basePath, namespace.String(), "permissions",
+	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "permissions",
 		"subscription", sName)
-	return n.client.post(endpoint, roles)
+	return n.pulsar.Client.Post(endpoint, roles)
 }
 
 func (n *namespaces) RevokeSubPermission(namespace utils.NameSpaceName, sName, role string) error {
-	endpoint := n.client.endpoint(n.basePath, namespace.String(), "permissions",
+	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "permissions",
 		"subscription", sName, role)
-	return n.client.delete(endpoint)
+	return n.pulsar.Client.Delete(endpoint)
 }
 
 func (n *namespaces) SetSubscriptionAuthMode(namespace utils.NameSpaceName, mode utils.SubscriptionAuthMode) error {
-	endpoint := n.client.endpoint(n.basePath, namespace.String(), "subscriptionAuthMode")
-	return n.client.post(endpoint, mode.String())
+	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "subscriptionAuthMode")
+	return n.pulsar.Client.Post(endpoint, mode.String())
 }
 
 func (n *namespaces) SetEncryptionRequiredStatus(namespace utils.NameSpaceName, encrypt bool) error {
-	endpoint := n.client.endpoint(n.basePath, namespace.String(), "encryptionRequired")
-	return n.client.post(endpoint, strconv.FormatBool(encrypt))
+	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "encryptionRequired")
+	return n.pulsar.Client.Post(endpoint, strconv.FormatBool(encrypt))
 }
 
 func (n *namespaces) UnsubscribeNamespace(namespace utils.NameSpaceName, sName string) error {
-	endpoint := n.client.endpoint(n.basePath, namespace.String(), "unsubscribe", url.QueryEscape(sName))
-	return n.client.post(endpoint, "")
+	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "unsubscribe", url.QueryEscape(sName))
+	return n.pulsar.Client.Post(endpoint, "")
 }
 
 func (n *namespaces) UnsubscribeNamespaceBundle(namespace utils.NameSpaceName, bundle, sName string) error {
-	endpoint := n.client.endpoint(n.basePath, namespace.String(), bundle, "unsubscribe", url.QueryEscape(sName))
-	return n.client.post(endpoint, "")
+	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), bundle, "unsubscribe", url.QueryEscape(sName))
+	return n.pulsar.Client.Post(endpoint, "")
 }
 
 func (n *namespaces) ClearNamespaceBundleBacklogForSubscription(namespace utils.NameSpaceName,
 	bundle, sName string) error {
-	endpoint := n.client.endpoint(n.basePath, namespace.String(), bundle, "clearBacklog", url.QueryEscape(sName))
-	return n.client.post(endpoint, "")
+	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), bundle, "clearBacklog", url.QueryEscape(sName))
+	return n.pulsar.Client.Post(endpoint, "")
 }
 
 func (n *namespaces) ClearNamespaceBundleBacklog(namespace utils.NameSpaceName, bundle string) error {
-	endpoint := n.client.endpoint(n.basePath, namespace.String(), bundle, "clearBacklog")
-	return n.client.post(endpoint, "")
+	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), bundle, "clearBacklog")
+	return n.pulsar.Client.Post(endpoint, "")
 }
 
 func (n *namespaces) ClearNamespaceBacklogForSubscription(namespace utils.NameSpaceName, sName string) error {
-	endpoint := n.client.endpoint(n.basePath, namespace.String(), "clearBacklog", url.QueryEscape(sName))
-	return n.client.post(endpoint, "")
+	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "clearBacklog", url.QueryEscape(sName))
+	return n.pulsar.Client.Post(endpoint, "")
 }
 
 func (n *namespaces) ClearNamespaceBacklog(namespace utils.NameSpaceName) error {
-	endpoint := n.client.endpoint(n.basePath, namespace.String(), "clearBacklog")
-	return n.client.post(endpoint, "")
+	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "clearBacklog")
+	return n.pulsar.Client.Post(endpoint, "")
 }
 
 func (n *namespaces) SetReplicatorDispatchRate(namespace utils.NameSpaceName, rate utils.DispatchRate) error {
-	endpoint := n.client.endpoint(n.basePath, namespace.String(), "replicatorDispatchRate")
-	return n.client.post(endpoint, rate)
+	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "replicatorDispatchRate")
+	return n.pulsar.Client.Post(endpoint, rate)
 }
 
 func (n *namespaces) GetReplicatorDispatchRate(namespace utils.NameSpaceName) (utils.DispatchRate, error) {
-	endpoint := n.client.endpoint(n.basePath, namespace.String(), "replicatorDispatchRate")
+	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "replicatorDispatchRate")
 	var rate utils.DispatchRate
-	err := n.client.get(endpoint, &rate)
+	err := n.pulsar.Client.Get(endpoint, &rate)
 	return rate, err
 }
 
 func (n *namespaces) SetSubscriptionDispatchRate(namespace utils.NameSpaceName, rate utils.DispatchRate) error {
-	endpoint := n.client.endpoint(n.basePath, namespace.String(), "subscriptionDispatchRate")
-	return n.client.post(endpoint, rate)
+	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "subscriptionDispatchRate")
+	return n.pulsar.Client.Post(endpoint, rate)
 }
 
 func (n *namespaces) GetSubscriptionDispatchRate(namespace utils.NameSpaceName) (utils.DispatchRate, error) {
-	endpoint := n.client.endpoint(n.basePath, namespace.String(), "subscriptionDispatchRate")
+	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "subscriptionDispatchRate")
 	var rate utils.DispatchRate
-	err := n.client.get(endpoint, &rate)
+	err := n.pulsar.Client.Get(endpoint, &rate)
 	return rate, err
 }
 
 func (n *namespaces) SetSubscribeRate(namespace utils.NameSpaceName, rate utils.SubscribeRate) error {
-	endpoint := n.client.endpoint(n.basePath, namespace.String(), "subscribeRate")
-	return n.client.post(endpoint, rate)
+	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "subscribeRate")
+	return n.pulsar.Client.Post(endpoint, rate)
 }
 
 func (n *namespaces) GetSubscribeRate(namespace utils.NameSpaceName) (utils.SubscribeRate, error) {
-	endpoint := n.client.endpoint(n.basePath, namespace.String(), "subscribeRate")
+	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "subscribeRate")
 	var rate utils.SubscribeRate
-	err := n.client.get(endpoint, &rate)
+	err := n.pulsar.Client.Get(endpoint, &rate)
 	return rate, err
 }
 
 func (n *namespaces) SetDispatchRate(namespace utils.NameSpaceName, rate utils.DispatchRate) error {
-	endpoint := n.client.endpoint(n.basePath, namespace.String(), "dispatchRate")
-	return n.client.post(endpoint, rate)
+	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "dispatchRate")
+	return n.pulsar.Client.Post(endpoint, rate)
 }
 
 func (n *namespaces) GetDispatchRate(namespace utils.NameSpaceName) (utils.DispatchRate, error) {
-	endpoint := n.client.endpoint(n.basePath, namespace.String(), "dispatchRate")
+	endpoint := n.pulsar.endpoint(n.basePath, namespace.String(), "dispatchRate")
 	var rate utils.DispatchRate
-	err := n.client.get(endpoint, &rate)
+	err := n.pulsar.Client.Get(endpoint, &rate)
 	return rate, err
 }
diff --git a/pkg/pulsar/ns_isolation_policy.go b/pkg/pulsar/ns_isolation_policy.go
index 6c976319d..9f3762ec7 100644
--- a/pkg/pulsar/ns_isolation_policy.go
+++ b/pkg/pulsar/ns_isolation_policy.go
@@ -42,13 +42,13 @@ type NsIsolationPolicy interface {
 }
 
 type nsIsolationPolicy struct {
-	client   *client
+	pulsar   *pulsarClient
 	basePath string
 }
 
-func (c *client) NsIsolationPolicy() NsIsolationPolicy {
+func (c *pulsarClient) NsIsolationPolicy() NsIsolationPolicy {
 	return &nsIsolationPolicy{
-		client:   c,
+		pulsar:   c,
 		basePath: "/clusters",
 	}
 }
@@ -60,20 +60,20 @@ func (n *nsIsolationPolicy) CreateNamespaceIsolationPolicy(cluster, policyName s
 
 func (n *nsIsolationPolicy) setNamespaceIsolationPolicy(cluster, policyName string,
 	namespaceIsolationData utils.NamespaceIsolationData) error {
-	endpoint := n.client.endpoint(n.basePath, cluster, "namespaceIsolationPolicies", policyName)
-	return n.client.post(endpoint, &namespaceIsolationData)
+	endpoint := n.pulsar.endpoint(n.basePath, cluster, "namespaceIsolationPolicies", policyName)
+	return n.pulsar.Client.Post(endpoint, &namespaceIsolationData)
 }
 
 func (n *nsIsolationPolicy) DeleteNamespaceIsolationPolicy(cluster, policyName string) error {
-	endpoint := n.client.endpoint(n.basePath, cluster, "namespaceIsolationPolicies", policyName)
-	return n.client.delete(endpoint)
+	endpoint := n.pulsar.endpoint(n.basePath, cluster, "namespaceIsolationPolicies", policyName)
+	return n.pulsar.Client.Delete(endpoint)
 }
 
 func (n *nsIsolationPolicy) GetNamespaceIsolationPolicy(cluster, policyName string) (
 	*utils.NamespaceIsolationData, error) {
-	endpoint := n.client.endpoint(n.basePath, cluster, "namespaceIsolationPolicies", policyName)
+	endpoint := n.pulsar.endpoint(n.basePath, cluster, "namespaceIsolationPolicies", policyName)
 	var nsIsolationData utils.NamespaceIsolationData
-	err := n.client.get(endpoint, &nsIsolationData)
+	err := n.pulsar.Client.Get(endpoint, &nsIsolationData)
 	if err != nil {
 		return nil, err
 	}
@@ -82,9 +82,9 @@ func (n *nsIsolationPolicy) GetNamespaceIsolationPolicy(cluster, policyName stri
 
 func (n *nsIsolationPolicy) GetNamespaceIsolationPolicies(cluster string) (
 	map[string]utils.NamespaceIsolationData, error) {
-	endpoint := n.client.endpoint(n.basePath, cluster, "namespaceIsolationPolicies")
+	endpoint := n.pulsar.endpoint(n.basePath, cluster, "namespaceIsolationPolicies")
 	var tmpMap map[string]utils.NamespaceIsolationData
-	err := n.client.get(endpoint, &tmpMap)
+	err := n.pulsar.Client.Get(endpoint, &tmpMap)
 	if err != nil {
 		return nil, err
 	}
@@ -93,9 +93,9 @@ func (n *nsIsolationPolicy) GetNamespaceIsolationPolicies(cluster string) (
 
 func (n *nsIsolationPolicy) GetBrokersWithNamespaceIsolationPolicy(cluster string) (
 	[]utils.BrokerNamespaceIsolationData, error) {
-	endpoint := n.client.endpoint(n.basePath, cluster, "namespaceIsolationPolicies", "brokers")
+	endpoint := n.pulsar.endpoint(n.basePath, cluster, "namespaceIsolationPolicies", "brokers")
 	var res []utils.BrokerNamespaceIsolationData
-	err := n.client.get(endpoint, &res)
+	err := n.pulsar.Client.Get(endpoint, &res)
 	if err != nil {
 		return nil, err
 	}
@@ -104,9 +104,9 @@ func (n *nsIsolationPolicy) GetBrokersWithNamespaceIsolationPolicy(cluster strin
 
 func (n *nsIsolationPolicy) GetBrokerWithNamespaceIsolationPolicy(cluster,
 	broker string) (*utils.BrokerNamespaceIsolationData, error) {
-	endpoint := n.client.endpoint(n.basePath, cluster, "namespaceIsolationPolicies", "brokers", broker)
+	endpoint := n.pulsar.endpoint(n.basePath, cluster, "namespaceIsolationPolicies", "brokers", broker)
 	var brokerNamespaceIsolationData utils.BrokerNamespaceIsolationData
-	err := n.client.get(endpoint, &brokerNamespaceIsolationData)
+	err := n.pulsar.Client.Get(endpoint, &brokerNamespaceIsolationData)
 	if err != nil {
 		return nil, err
 	}
diff --git a/pkg/pulsar/resource_quotas.go b/pkg/pulsar/resource_quotas.go
index de2873000..3fc0609c5 100644
--- a/pkg/pulsar/resource_quotas.go
+++ b/pkg/pulsar/resource_quotas.go
@@ -39,21 +39,21 @@ type ResourceQuotas interface {
 }
 
 type resource struct {
-	client   *client
+	pulsar   *pulsarClient
 	basePath string
 }
 
-func (c *client) ResourceQuotas() ResourceQuotas {
+func (c *pulsarClient) ResourceQuotas() ResourceQuotas {
 	return &resource{
-		client:   c,
+		pulsar:   c,
 		basePath: "/resource-quotas",
 	}
 }
 
 func (r *resource) GetDefaultResourceQuota() (*utils.ResourceQuota, error) {
-	endpoint := r.client.endpoint(r.basePath)
+	endpoint := r.pulsar.endpoint(r.basePath)
 	var quota utils.ResourceQuota
-	err := r.client.get(endpoint, &quota)
+	err := r.pulsar.Client.Get(endpoint, &quota)
 	if err != nil {
 		return nil, err
 	}
@@ -61,14 +61,14 @@ func (r *resource) GetDefaultResourceQuota() (*utils.ResourceQuota, error) {
 }
 
 func (r *resource) SetDefaultResourceQuota(quota utils.ResourceQuota) error {
-	endpoint := r.client.endpoint(r.basePath)
-	return r.client.post(endpoint, &quota)
+	endpoint := r.pulsar.endpoint(r.basePath)
+	return r.pulsar.Client.Post(endpoint, &quota)
 }
 
 func (r *resource) GetNamespaceBundleResourceQuota(namespace, bundle string) (*utils.ResourceQuota, error) {
-	endpoint := r.client.endpoint(r.basePath, namespace, bundle)
+	endpoint := r.pulsar.endpoint(r.basePath, namespace, bundle)
 	var quota utils.ResourceQuota
-	err := r.client.get(endpoint, &quota)
+	err := r.pulsar.Client.Get(endpoint, &quota)
 	if err != nil {
 		return nil, err
 	}
@@ -76,11 +76,11 @@ func (r *resource) GetNamespaceBundleResourceQuota(namespace, bundle string) (*u
 }
 
 func (r *resource) SetNamespaceBundleResourceQuota(namespace, bundle string, quota utils.ResourceQuota) error {
-	endpoint := r.client.endpoint(r.basePath, namespace, bundle)
-	return r.client.post(endpoint, &quota)
+	endpoint := r.pulsar.endpoint(r.basePath, namespace, bundle)
+	return r.pulsar.Client.Post(endpoint, &quota)
 }
 
 func (r *resource) ResetNamespaceBundleResourceQuota(namespace, bundle string) error {
-	endpoint := r.client.endpoint(r.basePath, namespace, bundle)
-	return r.client.delete(endpoint)
+	endpoint := r.pulsar.endpoint(r.basePath, namespace, bundle)
+	return r.pulsar.Client.Delete(endpoint)
 }
diff --git a/pkg/pulsar/schema.go b/pkg/pulsar/schema.go
index bdcba21fc..5fd57ec4a 100644
--- a/pkg/pulsar/schema.go
+++ b/pkg/pulsar/schema.go
@@ -43,14 +43,14 @@ type Schema interface {
 }
 
 type schemas struct {
-	client   *client
+	pulsar   *pulsarClient
 	basePath string
 }
 
 // Schemas is used to access the schemas endpoints
-func (c *client) Schemas() Schema {
+func (c *pulsarClient) Schemas() Schema {
 	return &schemas{
-		client:   c,
+		pulsar:   c,
 		basePath: "/schemas",
 	}
 }
@@ -61,10 +61,10 @@ func (s *schemas) GetSchemaInfo(topic string) (*utils.SchemaInfo, error) {
 		return nil, err
 	}
 	var response utils.GetSchemaResponse
-	endpoint := s.client.endpoint(s.basePath, topicName.GetTenant(), topicName.GetNamespace(),
+	endpoint := s.pulsar.endpoint(s.basePath, topicName.GetTenant(), topicName.GetNamespace(),
 		topicName.GetEncodedTopic(), "schema")
 
-	err = s.client.get(endpoint, &response)
+	err = s.pulsar.Client.Get(endpoint, &response)
 	if err != nil {
 		return nil, err
 	}
@@ -79,10 +79,10 @@ func (s *schemas) GetSchemaInfoWithVersion(topic string) (*utils.SchemaInfoWithV
 		return nil, err
 	}
 	var response utils.GetSchemaResponse
-	endpoint := s.client.endpoint(s.basePath, topicName.GetTenant(), topicName.GetNamespace(),
+	endpoint := s.pulsar.endpoint(s.basePath, topicName.GetTenant(), topicName.GetNamespace(),
 		topicName.GetEncodedTopic(), "schema")
 
-	err = s.client.get(endpoint, &response)
+	err = s.pulsar.Client.Get(endpoint, &response)
 	if err != nil {
 		fmt.Println("err:", err.Error())
 		return nil, err
@@ -99,10 +99,10 @@ func (s *schemas) GetSchemaInfoByVersion(topic string, version int64) (*utils.Sc
 	}
 
 	var response utils.GetSchemaResponse
-	endpoint := s.client.endpoint(s.basePath, topicName.GetTenant(), topicName.GetNamespace(), topicName.GetEncodedTopic(),
+	endpoint := s.pulsar.endpoint(s.basePath, topicName.GetTenant(), topicName.GetNamespace(), topicName.GetEncodedTopic(),
 		"schema", strconv.FormatInt(version, 10))
 
-	err = s.client.get(endpoint, &response)
+	err = s.pulsar.Client.Get(endpoint, &response)
 	if err != nil {
 		return nil, err
 	}
@@ -117,12 +117,12 @@ func (s *schemas) DeleteSchema(topic string) error {
 		return err
 	}
 
-	endpoint := s.client.endpoint(s.basePath, topicName.GetTenant(), topicName.GetNamespace(),
+	endpoint := s.pulsar.endpoint(s.basePath, topicName.GetTenant(), topicName.GetNamespace(),
 		topicName.GetEncodedTopic(), "schema")
 
 	fmt.Println(endpoint)
 
-	return s.client.delete(endpoint)
+	return s.pulsar.Client.Delete(endpoint)
 }
 
 func (s *schemas) CreateSchemaByPayload(topic string, schemaPayload utils.PostSchemaPayload) error {
@@ -131,8 +131,8 @@ func (s *schemas) CreateSchemaByPayload(topic string, schemaPayload utils.PostSc
 		return err
 	}
 
-	endpoint := s.client.endpoint(s.basePath, topicName.GetTenant(), topicName.GetNamespace(),
+	endpoint := s.pulsar.endpoint(s.basePath, topicName.GetTenant(), topicName.GetNamespace(),
 		topicName.GetEncodedTopic(), "schema")
 
-	return s.client.post(endpoint, &schemaPayload)
+	return s.pulsar.Client.Post(endpoint, &schemaPayload)
 }
diff --git a/pkg/pulsar/sinks.go b/pkg/pulsar/sinks.go
index 7aa0dc700..1e26debbb 100644
--- a/pkg/pulsar/sinks.go
+++ b/pkg/pulsar/sinks.go
@@ -86,14 +86,14 @@ type Sinks interface {
 }
 
 type sinks struct {
-	client   *client
+	pulsar   *pulsarClient
 	basePath string
 }
 
 // Sinks is used to access the sinks endpoints
-func (c *client) Sinks() Sinks {
+func (c *pulsarClient) Sinks() Sinks {
 	return &sinks{
-		client:   c,
+		pulsar:   c,
 		basePath: "/sinks",
 	}
 }
@@ -114,20 +114,20 @@ func (s *sinks) createTextFromFiled(w *multipart.Writer, value string) (io.Write
 
 func (s *sinks) ListSinks(tenant, namespace string) ([]string, error) {
 	var sinks []string
-	endpoint := s.client.endpoint(s.basePath, tenant, namespace)
-	err := s.client.get(endpoint, &sinks)
+	endpoint := s.pulsar.endpoint(s.basePath, tenant, namespace)
+	err := s.pulsar.Client.Get(endpoint, &sinks)
 	return sinks, err
 }
 
 func (s *sinks) GetSink(tenant, namespace, sink string) (utils.SinkConfig, error) {
 	var sinkConfig utils.SinkConfig
-	endpoint := s.client.endpoint(s.basePath, tenant, namespace, sink)
-	err := s.client.get(endpoint, &sinkConfig)
+	endpoint := s.pulsar.endpoint(s.basePath, tenant, namespace, sink)
+	err := s.pulsar.Client.Get(endpoint, &sinkConfig)
 	return sinkConfig, err
 }
 
 func (s *sinks) CreateSink(config *utils.SinkConfig, fileName string) error {
-	endpoint := s.client.endpoint(s.basePath, config.Tenant, config.Namespace, config.Name)
+	endpoint := s.pulsar.endpoint(s.basePath, config.Tenant, config.Namespace, config.Name)
 
 	// buffer to store our request as bytes
 	bodyBuf := bytes.NewBufferString("")
@@ -176,7 +176,7 @@ func (s *sinks) CreateSink(config *utils.SinkConfig, fileName string) error {
 	}
 
 	contentType := multiPartWriter.FormDataContentType()
-	err = s.client.postWithMultiPart(endpoint, nil, bodyBuf, contentType)
+	err = s.pulsar.Client.PostWithMultiPart(endpoint, nil, bodyBuf, contentType)
 	if err != nil {
 		return err
 	}
@@ -185,7 +185,7 @@ func (s *sinks) CreateSink(config *utils.SinkConfig, fileName string) error {
 }
 
 func (s *sinks) CreateSinkWithURL(config *utils.SinkConfig, pkgURL string) error {
-	endpoint := s.client.endpoint(s.basePath, config.Tenant, config.Namespace, config.Name)
+	endpoint := s.pulsar.endpoint(s.basePath, config.Tenant, config.Namespace, config.Name)
 	// buffer to store our request as bytes
 	bodyBuf := bytes.NewBufferString("")
 
@@ -221,7 +221,7 @@ func (s *sinks) CreateSinkWithURL(config *utils.SinkConfig, pkgURL string) error
 	}
 
 	contentType := multiPartWriter.FormDataContentType()
-	err = s.client.postWithMultiPart(endpoint, nil, bodyBuf, contentType)
+	err = s.pulsar.Client.PostWithMultiPart(endpoint, nil, bodyBuf, contentType)
 	if err != nil {
 		return err
 	}
@@ -230,7 +230,7 @@ func (s *sinks) CreateSinkWithURL(config *utils.SinkConfig, pkgURL string) error
 }
 
 func (s *sinks) UpdateSink(config *utils.SinkConfig, fileName string, updateOptions *utils.UpdateOptions) error {
-	endpoint := s.client.endpoint(s.basePath, config.Tenant, config.Namespace, config.Name)
+	endpoint := s.pulsar.endpoint(s.basePath, config.Tenant, config.Namespace, config.Name)
 	// buffer to store our request as bytes
 	bodyBuf := bytes.NewBufferString("")
 
@@ -296,7 +296,7 @@ func (s *sinks) UpdateSink(config *utils.SinkConfig, fileName string, updateOpti
 	}
 
 	contentType := multiPartWriter.FormDataContentType()
-	err = s.client.putWithMultiPart(endpoint, bodyBuf, contentType)
+	err = s.pulsar.Client.PutWithMultiPart(endpoint, bodyBuf, contentType)
 	if err != nil {
 		return err
 	}
@@ -305,7 +305,7 @@ func (s *sinks) UpdateSink(config *utils.SinkConfig, fileName string, updateOpti
 }
 
 func (s *sinks) UpdateSinkWithURL(config *utils.SinkConfig, pkgURL string, updateOptions *utils.UpdateOptions) error {
-	endpoint := s.client.endpoint(s.basePath, config.Tenant, config.Namespace, config.Name)
+	endpoint := s.pulsar.endpoint(s.basePath, config.Tenant, config.Namespace, config.Name)
 	// buffer to store our request as bytes
 	bodyBuf := bytes.NewBufferString("")
 
@@ -360,7 +360,7 @@ func (s *sinks) UpdateSinkWithURL(config *utils.SinkConfig, pkgURL string, updat
 	}
 
 	contentType := multiPartWriter.FormDataContentType()
-	err = s.client.putWithMultiPart(endpoint, bodyBuf, contentType)
+	err = s.pulsar.Client.PutWithMultiPart(endpoint, bodyBuf, contentType)
 	if err != nil {
 		return err
 	}
@@ -369,69 +369,69 @@ func (s *sinks) UpdateSinkWithURL(config *utils.SinkConfig, pkgURL string, updat
 }
 
 func (s *sinks) DeleteSink(tenant, namespace, sink string) error {
-	endpoint := s.client.endpoint(s.basePath, tenant, namespace, sink)
-	return s.client.delete(endpoint)
+	endpoint := s.pulsar.endpoint(s.basePath, tenant, namespace, sink)
+	return s.pulsar.Client.Delete(endpoint)
 }
 
 func (s *sinks) GetSinkStatus(tenant, namespace, sink string) (utils.SinkStatus, error) {
 	var sinkStatus utils.SinkStatus
-	endpoint := s.client.endpoint(s.basePath, tenant, namespace, sink)
-	err := s.client.get(endpoint+"/status", &sinkStatus)
+	endpoint := s.pulsar.endpoint(s.basePath, tenant, namespace, sink)
+	err := s.pulsar.Client.Get(endpoint+"/status", &sinkStatus)
 	return sinkStatus, err
 }
 
 func (s *sinks) GetSinkStatusWithID(tenant, namespace, sink string, id int) (utils.SinkInstanceStatusData, error) {
 	var sinkInstanceStatusData utils.SinkInstanceStatusData
 	instanceID := fmt.Sprintf("%d", id)
-	endpoint := s.client.endpoint(s.basePath, tenant, namespace, sink, instanceID)
-	err := s.client.get(endpoint+"/status", &sinkInstanceStatusData)
+	endpoint := s.pulsar.endpoint(s.basePath, tenant, namespace, sink, instanceID)
+	err := s.pulsar.Client.Get(endpoint+"/status", &sinkInstanceStatusData)
 	return sinkInstanceStatusData, err
 }
 
 func (s *sinks) RestartSink(tenant, namespace, sink string) error {
-	endpoint := s.client.endpoint(s.basePath, tenant, namespace, sink)
-	return s.client.post(endpoint+"/restart", "")
+	endpoint := s.pulsar.endpoint(s.basePath, tenant, namespace, sink)
+	return s.pulsar.Client.Post(endpoint+"/restart", "")
 }
 
 func (s *sinks) RestartSinkWithID(tenant, namespace, sink string, instanceID int) error {
 	id := fmt.Sprintf("%d", instanceID)
-	endpoint := s.client.endpoint(s.basePath, tenant, namespace, sink, id)
+	endpoint := s.pulsar.endpoint(s.basePath, tenant, namespace, sink, id)
 
-	return s.client.post(endpoint+"/restart", "")
+	return s.pulsar.Client.Post(endpoint+"/restart", "")
 }
 
 func (s *sinks) StopSink(tenant, namespace, sink string) error {
-	endpoint := s.client.endpoint(s.basePath, tenant, namespace, sink)
-	return s.client.post(endpoint+"/stop", "")
+	endpoint := s.pulsar.endpoint(s.basePath, tenant, namespace, sink)
+	return s.pulsar.Client.Post(endpoint+"/stop", "")
 }
 
 func (s *sinks) StopSinkWithID(tenant, namespace, sink string, instanceID int) error {
 	id := fmt.Sprintf("%d", instanceID)
-	endpoint := s.client.endpoint(s.basePath, tenant, namespace, sink, id)
+	endpoint := s.pulsar.endpoint(s.basePath, tenant, namespace, sink, id)
 
-	return s.client.post(endpoint+"/stop", "")
+	return s.pulsar.Client.Post(endpoint+"/stop", "")
 }
 
 func (s *sinks) StartSink(tenant, namespace, sink string) error {
-	endpoint := s.client.endpoint(s.basePath, tenant, namespace, sink)
-	return s.client.post(endpoint+"/start", "")
+	endpoint := s.pulsar.endpoint(s.basePath, tenant, namespace, sink)
+	return s.pulsar.Client.Post(endpoint+"/start", "")
 }
 
 func (s *sinks) StartSinkWithID(tenant, namespace, sink string, instanceID int) error {
 	id := fmt.Sprintf("%d", instanceID)
-	endpoint := s.client.endpoint(s.basePath, tenant, namespace, sink, id)
+	endpoint := s.pulsar.endpoint(s.basePath, tenant, namespace, sink, id)
 
-	return s.client.post(endpoint+"/start", "")
+	return s.pulsar.Client.Post(endpoint+"/start", "")
 }
 
 func (s *sinks) GetBuiltInSinks() ([]*utils.ConnectorDefinition, error) {
 	var connectorDefinition []*utils.ConnectorDefinition
-	endpoint := s.client.endpoint(s.basePath, "builtinSinks")
-	err := s.client.get(endpoint, &connectorDefinition)
+	endpoint := s.pulsar.endpoint(s.basePath, "builtinSinks")
+	err := s.pulsar.Client.Get(endpoint, &connectorDefinition)
 	return connectorDefinition, err
 }
 
 func (s *sinks) ReloadBuiltInSinks() error {
-	endpoint := s.client.endpoint(s.basePath, "reloadBuiltInSinks")
-	return s.client.post(endpoint, "")
+	endpoint := s.pulsar.endpoint(s.basePath, "reloadBuiltInSinks")
+	return s.pulsar.Client.Post(endpoint, "")
 }
diff --git a/pkg/pulsar/sources.go b/pkg/pulsar/sources.go
index 3119d6d23..7fabfacaf 100644
--- a/pkg/pulsar/sources.go
+++ b/pkg/pulsar/sources.go
@@ -87,14 +87,14 @@ type Sources interface {
 }
 
 type sources struct {
-	client   *client
+	pulsar   *pulsarClient
 	basePath string
 }
 
 // Sources is used to access the sources endpoints
-func (c *client) Sources() Sources {
+func (c *pulsarClient) Sources() Sources {
 	return &sources{
-		client:   c,
+		pulsar:   c,
 		basePath: "/sources",
 	}
 }
@@ -115,20 +115,20 @@ func (s *sources) createTextFromFiled(w *multipart.Writer, value string) (io.Wri
 
 func (s *sources) ListSources(tenant, namespace string) ([]string, error) {
 	var sources []string
-	endpoint := s.client.endpoint(s.basePath, tenant, namespace)
-	err := s.client.get(endpoint, &sources)
+	endpoint := s.pulsar.endpoint(s.basePath, tenant, namespace)
+	err := s.pulsar.Client.Get(endpoint, &sources)
 	return sources, err
 }
 
 func (s *sources) GetSource(tenant, namespace, source string) (utils.SourceConfig, error) {
 	var sourceConfig utils.SourceConfig
-	endpoint := s.client.endpoint(s.basePath, tenant, namespace, source)
-	err := s.client.get(endpoint, &sourceConfig)
+	endpoint := s.pulsar.endpoint(s.basePath, tenant, namespace, source)
+	err := s.pulsar.Client.Get(endpoint, &sourceConfig)
 	return sourceConfig, err
 }
 
 func (s *sources) CreateSource(config *utils.SourceConfig, fileName string) error {
-	endpoint := s.client.endpoint(s.basePath, config.Tenant, config.Namespace, config.Name)
+	endpoint := s.pulsar.endpoint(s.basePath, config.Tenant, config.Namespace, config.Name)
 
 	// buffer to store our request as bytes
 	bodyBuf := bytes.NewBufferString("")
@@ -177,7 +177,7 @@ func (s *sources) CreateSource(config *utils.SourceConfig, fileName string) erro
 	}
 
 	contentType := multiPartWriter.FormDataContentType()
-	err = s.client.postWithMultiPart(endpoint, nil, bodyBuf, contentType)
+	err = s.pulsar.Client.PostWithMultiPart(endpoint, nil, bodyBuf, contentType)
 	if err != nil {
 		return err
 	}
@@ -186,7 +186,7 @@ func (s *sources) CreateSource(config *utils.SourceConfig, fileName string) erro
 }
 
 func (s *sources) CreateSourceWithURL(config *utils.SourceConfig, pkgURL string) error {
-	endpoint := s.client.endpoint(s.basePath, config.Tenant, config.Namespace, config.Name)
+	endpoint := s.pulsar.endpoint(s.basePath, config.Tenant, config.Namespace, config.Name)
 	// buffer to store our request as bytes
 	bodyBuf := bytes.NewBufferString("")
 
@@ -222,7 +222,7 @@ func (s *sources) CreateSourceWithURL(config *utils.SourceConfig, pkgURL string)
 	}
 
 	contentType := multiPartWriter.FormDataContentType()
-	err = s.client.postWithMultiPart(endpoint, nil, bodyBuf, contentType)
+	err = s.pulsar.Client.PostWithMultiPart(endpoint, nil, bodyBuf, contentType)
 	if err != nil {
 		return err
 	}
@@ -231,7 +231,7 @@ func (s *sources) CreateSourceWithURL(config *utils.SourceConfig, pkgURL string)
 }
 
 func (s *sources) UpdateSource(config *utils.SourceConfig, fileName string, updateOptions *utils.UpdateOptions) error {
-	endpoint := s.client.endpoint(s.basePath, config.Tenant, config.Namespace, config.Name)
+	endpoint := s.pulsar.endpoint(s.basePath, config.Tenant, config.Namespace, config.Name)
 	// buffer to store our request as bytes
 	bodyBuf := bytes.NewBufferString("")
 
@@ -297,7 +297,7 @@ func (s *sources) UpdateSource(config *utils.SourceConfig, fileName string, upda
 	}
 
 	contentType := multiPartWriter.FormDataContentType()
-	err = s.client.putWithMultiPart(endpoint, bodyBuf, contentType)
+	err = s.pulsar.Client.PutWithMultiPart(endpoint, bodyBuf, contentType)
 	if err != nil {
 		return err
 	}
@@ -307,7 +307,7 @@ func (s *sources) UpdateSource(config *utils.SourceConfig, fileName string, upda
 
 func (s *sources) UpdateSourceWithURL(config *utils.SourceConfig, pkgURL string,
 	updateOptions *utils.UpdateOptions) error {
-	endpoint := s.client.endpoint(s.basePath, config.Tenant, config.Namespace, config.Name)
+	endpoint := s.pulsar.endpoint(s.basePath, config.Tenant, config.Namespace, config.Name)
 	// buffer to store our request as bytes
 	bodyBuf := bytes.NewBufferString("")
 
@@ -362,7 +362,7 @@ func (s *sources) UpdateSourceWithURL(config *utils.SourceConfig, pkgURL string,
 	}
 
 	contentType := multiPartWriter.FormDataContentType()
-	err = s.client.putWithMultiPart(endpoint, bodyBuf, contentType)
+	err = s.pulsar.Client.PutWithMultiPart(endpoint, bodyBuf, contentType)
 	if err != nil {
 		return err
 	}
@@ -371,14 +371,14 @@ func (s *sources) UpdateSourceWithURL(config *utils.SourceConfig, pkgURL string,
 }
 
 func (s *sources) DeleteSource(tenant, namespace, source string) error {
-	endpoint := s.client.endpoint(s.basePath, tenant, namespace, source)
-	return s.client.delete(endpoint)
+	endpoint := s.pulsar.endpoint(s.basePath, tenant, namespace, source)
+	return s.pulsar.Client.Delete(endpoint)
 }
 
 func (s *sources) GetSourceStatus(tenant, namespace, source string) (utils.SourceStatus, error) {
 	var sourceStatus utils.SourceStatus
-	endpoint := s.client.endpoint(s.basePath, tenant, namespace, source)
-	err := s.client.get(endpoint+"/status", &sourceStatus)
+	endpoint := s.pulsar.endpoint(s.basePath, tenant, namespace, source)
+	err := s.pulsar.Client.Get(endpoint+"/status", &sourceStatus)
 	return sourceStatus, err
 }
 
@@ -386,55 +386,55 @@ func (s *sources) GetSourceStatusWithID(tenant, namespace, source string, id int
 	utils.SourceInstanceStatusData, error) {
 	var sourceInstanceStatusData utils.SourceInstanceStatusData
 	instanceID := fmt.Sprintf("%d", id)
-	endpoint := s.client.endpoint(s.basePath, tenant, namespace, source, instanceID)
-	err := s.client.get(endpoint+"/status", &sourceInstanceStatusData)
+	endpoint := s.pulsar.endpoint(s.basePath, tenant, namespace, source, instanceID)
+	err := s.pulsar.Client.Get(endpoint+"/status", &sourceInstanceStatusData)
 	return sourceInstanceStatusData, err
 }
 
 func (s *sources) RestartSource(tenant, namespace, source string) error {
-	endpoint := s.client.endpoint(s.basePath, tenant, namespace, source)
-	return s.client.post(endpoint+"/restart", "")
+	endpoint := s.pulsar.endpoint(s.basePath, tenant, namespace, source)
+	return s.pulsar.Client.Post(endpoint+"/restart", "")
 }
 
 func (s *sources) RestartSourceWithID(tenant, namespace, source string, instanceID int) error {
 	id := fmt.Sprintf("%d", instanceID)
-	endpoint := s.client.endpoint(s.basePath, tenant, namespace, source, id)
+	endpoint := s.pulsar.endpoint(s.basePath, tenant, namespace, source, id)
 
-	return s.client.post(endpoint+"/restart", "")
+	return s.pulsar.Client.Post(endpoint+"/restart", "")
 }
 
 func (s *sources) StopSource(tenant, namespace, source string) error {
-	endpoint := s.client.endpoint(s.basePath, tenant, namespace, source)
-	return s.client.post(endpoint+"/stop", "")
+	endpoint := s.pulsar.endpoint(s.basePath, tenant, namespace, source)
+	return s.pulsar.Client.Post(endpoint+"/stop", "")
 }
 
 func (s *sources) StopSourceWithID(tenant, namespace, source string, instanceID int) error {
 	id := fmt.Sprintf("%d", instanceID)
-	endpoint := s.client.endpoint(s.basePath, tenant, namespace, source, id)
+	endpoint := s.pulsar.endpoint(s.basePath, tenant, namespace, source, id)
 
-	return s.client.post(endpoint+"/stop", "")
+	return s.pulsar.Client.Post(endpoint+"/stop", "")
 }
 
 func (s *sources) StartSource(tenant, namespace, source string) error {
-	endpoint := s.client.endpoint(s.basePath, tenant, namespace, source)
-	return s.client.post(endpoint+"/start", "")
+	endpoint := s.pulsar.endpoint(s.basePath, tenant, namespace, source)
+	return s.pulsar.Client.Post(endpoint+"/start", "")
 }
 
 func (s *sources) StartSourceWithID(tenant, namespace, source string, instanceID int) error {
 	id := fmt.Sprintf("%d", instanceID)
-	endpoint := s.client.endpoint(s.basePath, tenant, namespace, source, id)
+	endpoint := s.pulsar.endpoint(s.basePath, tenant, namespace, source, id)
 
-	return s.client.post(endpoint+"/start", "")
+	return s.pulsar.Client.Post(endpoint+"/start", "")
 }
 
 func (s *sources) GetBuiltInSources() ([]*utils.ConnectorDefinition, error) {
 	var connectorDefinition []*utils.ConnectorDefinition
-	endpoint := s.client.endpoint(s.basePath, "builtinsources")
-	err := s.client.get(endpoint, &connectorDefinition)
+	endpoint := s.pulsar.endpoint(s.basePath, "builtinsources")
+	err := s.pulsar.Client.Get(endpoint, &connectorDefinition)
 	return connectorDefinition, err
 }
 
 func (s *sources) ReloadBuiltInSources() error {
-	endpoint := s.client.endpoint(s.basePath, "reloadBuiltInSources")
-	return s.client.post(endpoint, "")
+	endpoint := s.pulsar.endpoint(s.basePath, "reloadBuiltInSources")
+	return s.pulsar.Client.Post(endpoint, "")
 }
diff --git a/pkg/pulsar/subscription.go b/pkg/pulsar/subscription.go
index 37b94d7e2..143e33411 100644
--- a/pkg/pulsar/subscription.go
+++ b/pkg/pulsar/subscription.go
@@ -71,73 +71,73 @@ type Subscriptions interface {
 }
 
 type subscriptions struct {
-	client   *client
+	pulsar   *pulsarClient
 	basePath string
 	SubPath  string
 }
 
 // Subscriptions is used to access the subscriptions endpoints
-func (c *client) Subscriptions() Subscriptions {
+func (c *pulsarClient) Subscriptions() Subscriptions {
 	return &subscriptions{
-		client:   c,
+		pulsar:   c,
 		basePath: "",
 		SubPath:  "subscription",
 	}
 }
 
 func (s *subscriptions) Create(topic utils.TopicName, sName string, messageID utils.MessageID) error {
-	endpoint := s.client.endpoint(s.basePath, topic.GetRestPath(), s.SubPath, url.QueryEscape(sName))
-	return s.client.put(endpoint, messageID)
+	endpoint := s.pulsar.endpoint(s.basePath, topic.GetRestPath(), s.SubPath, url.QueryEscape(sName))
+	return s.pulsar.Client.Put(endpoint, messageID)
 }
 
 func (s *subscriptions) Delete(topic utils.TopicName, sName string) error {
-	endpoint := s.client.endpoint(s.basePath, topic.GetRestPath(), s.SubPath, url.QueryEscape(sName))
-	return s.client.delete(endpoint)
+	endpoint := s.pulsar.endpoint(s.basePath, topic.GetRestPath(), s.SubPath, url.QueryEscape(sName))
+	return s.pulsar.Client.Delete(endpoint)
 }
 
 func (s *subscriptions) List(topic utils.TopicName) ([]string, error) {
-	endpoint := s.client.endpoint(s.basePath, topic.GetRestPath(), "subscriptions")
+	endpoint := s.pulsar.endpoint(s.basePath, topic.GetRestPath(), "subscriptions")
 	var list []string
-	return list, s.client.get(endpoint, &list)
+	return list, s.pulsar.Client.Get(endpoint, &list)
 }
 
 func (s *subscriptions) ResetCursorToMessageID(topic utils.TopicName, sName string, id utils.MessageID) error {
-	endpoint := s.client.endpoint(s.basePath, topic.GetRestPath(), s.SubPath, url.QueryEscape(sName), "resetcursor")
-	return s.client.post(endpoint, id)
+	endpoint := s.pulsar.endpoint(s.basePath, topic.GetRestPath(), s.SubPath, url.QueryEscape(sName), "resetcursor")
+	return s.pulsar.Client.Post(endpoint, id)
 }
 
 func (s *subscriptions) ResetCursorToTimestamp(topic utils.TopicName, sName string, timestamp int64) error {
-	endpoint := s.client.endpoint(
+	endpoint := s.pulsar.endpoint(
 		s.basePath, topic.GetRestPath(), s.SubPath, url.QueryEscape(sName),
 		"resetcursor", strconv.FormatInt(timestamp, 10))
-	return s.client.post(endpoint, "")
+	return s.pulsar.Client.Post(endpoint, "")
 }
 
 func (s *subscriptions) ClearBacklog(topic utils.TopicName, sName string) error {
-	endpoint := s.client.endpoint(
+	endpoint := s.pulsar.endpoint(
 		s.basePath, topic.GetRestPath(), s.SubPath, url.QueryEscape(sName), "skip_all")
-	return s.client.post(endpoint, "")
+	return s.pulsar.Client.Post(endpoint, "")
 }
 
 func (s *subscriptions) SkipMessages(topic utils.TopicName, sName string, n int64) error {
-	endpoint := s.client.endpoint(
+	endpoint := s.pulsar.endpoint(
 		s.basePath, topic.GetRestPath(), s.SubPath, url.QueryEscape(sName),
 		"skip", strconv.FormatInt(n, 10))
-	return s.client.post(endpoint, "")
+	return s.pulsar.Client.Post(endpoint, "")
 }
 
 func (s *subscriptions) ExpireMessages(topic utils.TopicName, sName string, expire int64) error {
-	endpoint := s.client.endpoint(
+	endpoint := s.pulsar.endpoint(
 		s.basePath, topic.GetRestPath(), s.SubPath, url.QueryEscape(sName),
 		"expireMessages", strconv.FormatInt(expire, 10))
-	return s.client.post(endpoint, "")
+	return s.pulsar.Client.Post(endpoint, "")
 }
 
 func (s *subscriptions) ExpireAllMessages(topic utils.TopicName, expire int64) error {
-	endpoint := s.client.endpoint(
+	endpoint := s.pulsar.endpoint(
 		s.basePath, topic.GetRestPath(), "all_subscription",
 		"expireMessages", strconv.FormatInt(expire, 10))
-	return s.client.post(endpoint, "")
+	return s.pulsar.Client.Post(endpoint, "")
 }
 
 func (s *subscriptions) PeekMessages(topic utils.TopicName, sName string, n int) ([]*utils.Message, error) {
@@ -158,14 +158,10 @@ func (s *subscriptions) PeekMessages(topic utils.TopicName, sName string, n int)
 }
 
 func (s *subscriptions) peekNthMessage(topic utils.TopicName, sName string, pos int) ([]*utils.Message, error) {
-	endpoint := s.client.endpoint(s.basePath, topic.GetRestPath(), "subscription", url.QueryEscape(sName),
+	endpoint := s.pulsar.endpoint(s.basePath, topic.GetRestPath(), "subscription", url.QueryEscape(sName),
 		"position", strconv.Itoa(pos))
-	req, err := s.client.newRequest(http.MethodGet, endpoint)
-	if err != nil {
-		return nil, err
-	}
 
-	resp, err := checkSuccessful(s.client.doRequest(req))
+	resp, err := s.pulsar.Client.MakeRequest(http.MethodGet, endpoint)
 	if err != nil {
 		return nil, err
 	}
@@ -174,6 +170,14 @@ func (s *subscriptions) peekNthMessage(topic utils.TopicName, sName string, pos
 	return handleResp(topic, resp)
 }
 
+// safeRespClose is used to close a response body
+func safeRespClose(resp *http.Response) {
+	if resp != nil {
+		// ignore error since it is closing a response body
+		_ = resp.Body.Close()
+	}
+}
+
 const (
 	PublishTimeHeader = "X-Pulsar-Publish-Time"
 	BatchHeader       = "X-Pulsar-Num-Batch-Message"
diff --git a/pkg/pulsar/tenant.go b/pkg/pulsar/tenant.go
index c91adae2e..1d85d97da 100644
--- a/pkg/pulsar/tenant.go
+++ b/pkg/pulsar/tenant.go
@@ -40,43 +40,43 @@ type Tenants interface {
 }
 
 type tenants struct {
-	client   *client
+	pulsar   *pulsarClient
 	basePath string
 }
 
 // Tenants is used to access the tenants endpoints
-func (c *client) Tenants() Tenants {
+func (c *pulsarClient) Tenants() Tenants {
 	return &tenants{
-		client:   c,
+		pulsar:   c,
 		basePath: "/tenants",
 	}
 }
 
 func (c *tenants) Create(data utils.TenantData) error {
-	endpoint := c.client.endpoint(c.basePath, data.Name)
-	return c.client.put(endpoint, &data)
+	endpoint := c.pulsar.endpoint(c.basePath, data.Name)
+	return c.pulsar.Client.Put(endpoint, &data)
 }
 
 func (c *tenants) Delete(name string) error {
-	endpoint := c.client.endpoint(c.basePath, name)
-	return c.client.delete(endpoint)
+	endpoint := c.pulsar.endpoint(c.basePath, name)
+	return c.pulsar.Client.Delete(endpoint)
 }
 
 func (c *tenants) Update(data utils.TenantData) error {
-	endpoint := c.client.endpoint(c.basePath, data.Name)
-	return c.client.post(endpoint, &data)
+	endpoint := c.pulsar.endpoint(c.basePath, data.Name)
+	return c.pulsar.Client.Post(endpoint, &data)
 }
 
 func (c *tenants) List() ([]string, error) {
 	var tenantList []string
-	endpoint := c.client.endpoint(c.basePath, "")
-	err := c.client.get(endpoint, &tenantList)
+	endpoint := c.pulsar.endpoint(c.basePath, "")
+	err := c.pulsar.Client.Get(endpoint, &tenantList)
 	return tenantList, err
 }
 
 func (c *tenants) Get(name string) (utils.TenantData, error) {
 	var data utils.TenantData
-	endpoint := c.client.endpoint(c.basePath, name)
-	err := c.client.get(endpoint, &data)
+	endpoint := c.pulsar.endpoint(c.basePath, name)
+	err := c.pulsar.Client.Get(endpoint, &data)
 	return data, err
 }
diff --git a/pkg/pulsar/topic.go b/pkg/pulsar/topic.go
index 81dfd0ab0..24426514a 100644
--- a/pkg/pulsar/topic.go
+++ b/pkg/pulsar/topic.go
@@ -101,7 +101,7 @@ type Topics interface {
 }
 
 type topics struct {
-	client            *client
+	pulsar            *pulsarClient
 	basePath          string
 	persistentPath    string
 	nonPersistentPath string
@@ -109,9 +109,9 @@ type topics struct {
 }
 
 // Topics is used to access the topics endpoints
-func (c *client) Topics() Topics {
+func (c *pulsarClient) Topics() Topics {
 	return &topics{
-		client:            c,
+		pulsar:            c,
 		basePath:          "",
 		persistentPath:    "/persistent",
 		nonPersistentPath: "/non-persistent",
@@ -120,33 +120,33 @@ func (c *client) Topics() Topics {
 }
 
 func (t *topics) Create(topic utils.TopicName, partitions int) error {
-	endpoint := t.client.endpoint(t.basePath, topic.GetRestPath(), "partitions")
+	endpoint := t.pulsar.endpoint(t.basePath, topic.GetRestPath(), "partitions")
 	if partitions == 0 {
-		endpoint = t.client.endpoint(t.basePath, topic.GetRestPath())
+		endpoint = t.pulsar.endpoint(t.basePath, topic.GetRestPath())
 	}
-	return t.client.put(endpoint, partitions)
+	return t.pulsar.Client.Put(endpoint, partitions)
 }
 
 func (t *topics) Delete(topic utils.TopicName, force bool, nonPartitioned bool) error {
-	endpoint := t.client.endpoint(t.basePath, topic.GetRestPath(), "partitions")
+	endpoint := t.pulsar.endpoint(t.basePath, topic.GetRestPath(), "partitions")
 	if nonPartitioned {
-		endpoint = t.client.endpoint(t.basePath, topic.GetRestPath())
+		endpoint = t.pulsar.endpoint(t.basePath, topic.GetRestPath())
 	}
 	params := map[string]string{
 		"force": strconv.FormatBool(force),
 	}
-	return t.client.deleteWithQueryParams(endpoint, nil, params)
+	return t.pulsar.Client.DeleteWithQueryParams(endpoint, params)
 }
 
 func (t *topics) Update(topic utils.TopicName, partitions int) error {
-	endpoint := t.client.endpoint(t.basePath, topic.GetRestPath(), "partitions")
-	return t.client.post(endpoint, partitions)
+	endpoint := t.pulsar.endpoint(t.basePath, topic.GetRestPath(), "partitions")
+	return t.pulsar.Client.Post(endpoint, partitions)
 }
 
 func (t *topics) GetMetadata(topic utils.TopicName) (utils.PartitionedTopicMetadata, error) {
-	endpoint := t.client.endpoint(t.basePath, topic.GetRestPath(), "partitions")
+	endpoint := t.pulsar.endpoint(t.basePath, topic.GetRestPath(), "partitions")
 	var partitionedMeta utils.PartitionedTopicMetadata
-	err := t.client.get(endpoint, &partitionedMeta)
+	err := t.pulsar.Client.Get(endpoint, &partitionedMeta)
 	return partitionedMeta, err
 }
 
@@ -156,10 +156,10 @@ func (t *topics) List(namespace utils.NameSpaceName) ([]string, []string, error)
 	nonPartitionedTopicsChan := make(chan []string)
 	errChan := make(chan error)
 
-	pp := t.client.endpoint(t.persistentPath, namespace.String(), "partitioned")
-	np := t.client.endpoint(t.nonPersistentPath, namespace.String(), "partitioned")
-	p := t.client.endpoint(t.persistentPath, namespace.String())
-	n := t.client.endpoint(t.nonPersistentPath, namespace.String())
+	pp := t.pulsar.endpoint(t.persistentPath, namespace.String(), "partitioned")
+	np := t.pulsar.endpoint(t.nonPersistentPath, namespace.String(), "partitioned")
+	p := t.pulsar.endpoint(t.persistentPath, namespace.String())
+	n := t.pulsar.endpoint(t.nonPersistentPath, namespace.String())
 
 	go t.getTopics(pp, partitionedTopicsChan, errChan)
 	go t.getTopics(np, partitionedTopicsChan, errChan)
@@ -190,114 +190,114 @@ func (t *topics) List(namespace utils.NameSpaceName) ([]string, []string, error)
 
 func (t *topics) getTopics(endpoint string, out chan<- []string, err chan<- error) {
 	var topics []string
-	err <- t.client.get(endpoint, &topics)
+	err <- t.pulsar.Client.Get(endpoint, &topics)
 	out <- topics
 }
 
 func (t *topics) GetInternalInfo(topic utils.TopicName) (utils.ManagedLedgerInfo, error) {
-	endpoint := t.client.endpoint(t.basePath, topic.GetRestPath(), "internal-info")
+	endpoint := t.pulsar.endpoint(t.basePath, topic.GetRestPath(), "internal-info")
 	var info utils.ManagedLedgerInfo
-	err := t.client.get(endpoint, &info)
+	err := t.pulsar.Client.Get(endpoint, &info)
 	return info, err
 }
 
 func (t *topics) GetPermissions(topic utils.TopicName) (map[string][]common.AuthAction, error) {
 	var permissions map[string][]common.AuthAction
-	endpoint := t.client.endpoint(t.basePath, topic.GetRestPath(), "permissions")
-	err := t.client.get(endpoint, &permissions)
+	endpoint := t.pulsar.endpoint(t.basePath, topic.GetRestPath(), "permissions")
+	err := t.pulsar.Client.Get(endpoint, &permissions)
 	return permissions, err
 }
 
 func (t *topics) GrantPermission(topic utils.TopicName, role string, action []common.AuthAction) error {
-	endpoint := t.client.endpoint(t.basePath, topic.GetRestPath(), "permissions", role)
+	endpoint := t.pulsar.endpoint(t.basePath, topic.GetRestPath(), "permissions", role)
 	s := []string{}
 	for _, v := range action {
 		s = append(s, v.String())
 	}
-	return t.client.post(endpoint, s)
+	return t.pulsar.Client.Post(endpoint, s)
 }
 
 func (t *topics) RevokePermission(topic utils.TopicName, role string) error {
-	endpoint := t.client.endpoint(t.basePath, topic.GetRestPath(), "permissions", role)
-	return t.client.delete(endpoint)
+	endpoint := t.pulsar.endpoint(t.basePath, topic.GetRestPath(), "permissions", role)
+	return t.pulsar.Client.Delete(endpoint)
 }
 
 func (t *topics) Lookup(topic utils.TopicName) (utils.LookupData, error) {
 	var lookup utils.LookupData
 	endpoint := fmt.Sprintf("%s/%s", t.lookupPath, topic.GetRestPath())
-	err := t.client.get(endpoint, &lookup)
+	err := t.pulsar.Client.Get(endpoint, &lookup)
 	return lookup, err
 }
 
 func (t *topics) GetBundleRange(topic utils.TopicName) (string, error) {
 	endpoint := fmt.Sprintf("%s/%s/%s", t.lookupPath, topic.GetRestPath(), "bundle")
-	data, err := t.client.getWithQueryParams(endpoint, nil, nil, false)
+	data, err := t.pulsar.Client.GetWithQueryParams(endpoint, nil, nil, false)
 	return string(data), err
 }
 
 func (t *topics) GetLastMessageID(topic utils.TopicName) (utils.MessageID, error) {
 	var messageID utils.MessageID
-	endpoint := t.client.endpoint(t.basePath, topic.GetRestPath(), "lastMessageId")
-	err := t.client.get(endpoint, &messageID)
+	endpoint := t.pulsar.endpoint(t.basePath, topic.GetRestPath(), "lastMessageId")
+	err := t.pulsar.Client.Get(endpoint, &messageID)
 	return messageID, err
 }
 
 func (t *topics) GetStats(topic utils.TopicName) (utils.TopicStats, error) {
 	var stats utils.TopicStats
-	endpoint := t.client.endpoint(t.basePath, topic.GetRestPath(), "stats")
-	err := t.client.get(endpoint, &stats)
+	endpoint := t.pulsar.endpoint(t.basePath, topic.GetRestPath(), "stats")
+	err := t.pulsar.Client.Get(endpoint, &stats)
 	return stats, err
 }
 
 func (t *topics) GetInternalStats(topic utils.TopicName) (utils.PersistentTopicInternalStats, error) {
 	var stats utils.PersistentTopicInternalStats
-	endpoint := t.client.endpoint(t.basePath, topic.GetRestPath(), "internalStats")
-	err := t.client.get(endpoint, &stats)
+	endpoint := t.pulsar.endpoint(t.basePath, topic.GetRestPath(), "internalStats")
+	err := t.pulsar.Client.Get(endpoint, &stats)
 	return stats, err
 }
 
 func (t *topics) GetPartitionedStats(topic utils.TopicName, perPartition bool) (utils.PartitionedTopicStats, error) {
 	var stats utils.PartitionedTopicStats
-	endpoint := t.client.endpoint(t.basePath, topic.GetRestPath(), "partitioned-stats")
+	endpoint := t.pulsar.endpoint(t.basePath, topic.GetRestPath(), "partitioned-stats")
 	params := map[string]string{
 		"perPartition": strconv.FormatBool(perPartition),
 	}
-	_, err := t.client.getWithQueryParams(endpoint, &stats, params, true)
+	_, err := t.pulsar.Client.GetWithQueryParams(endpoint, &stats, params, true)
 	return stats, err
 }
 
 func (t *topics) Terminate(topic utils.TopicName) (utils.MessageID, error) {
-	endpoint := t.client.endpoint(t.basePath, topic.GetRestPath(), "terminate")
+	endpoint := t.pulsar.endpoint(t.basePath, topic.GetRestPath(), "terminate")
 	var messageID utils.MessageID
-	err := t.client.postWithObj(endpoint, "", &messageID)
+	err := t.pulsar.Client.PostWithObj(endpoint, "", &messageID)
 	return messageID, err
 }
 
 func (t *topics) Offload(topic utils.TopicName, messageID utils.MessageID) error {
-	endpoint := t.client.endpoint(t.basePath, topic.GetRestPath(), "offload")
-	return t.client.put(endpoint, messageID)
+	endpoint := t.pulsar.endpoint(t.basePath, topic.GetRestPath(), "offload")
+	return t.pulsar.Client.Put(endpoint, messageID)
 }
 
 func (t *topics) OffloadStatus(topic utils.TopicName) (utils.OffloadProcessStatus, error) {
-	endpoint := t.client.endpoint(t.basePath, topic.GetRestPath(), "offload")
+	endpoint := t.pulsar.endpoint(t.basePath, topic.GetRestPath(), "offload")
 	var status utils.OffloadProcessStatus
-	err := t.client.get(endpoint, &status)
+	err := t.pulsar.Client.Get(endpoint, &status)
 	return status, err
 }
 
 func (t *topics) Unload(topic utils.TopicName) error {
-	endpoint := t.client.endpoint(t.basePath, topic.GetRestPath(), "unload")
-	return t.client.put(endpoint, "")
+	endpoint := t.pulsar.endpoint(t.basePath, topic.GetRestPath(), "unload")
+	return t.pulsar.Client.Put(endpoint, "")
 }
 
 func (t *topics) Compact(topic utils.TopicName) error {
-	endpoint := t.client.endpoint(t.basePath, topic.GetRestPath(), "compaction")
-	return t.client.put(endpoint, "")
+	endpoint := t.pulsar.endpoint(t.basePath, topic.GetRestPath(), "compaction")
+	return t.pulsar.Client.Put(endpoint, "")
 }
 
 func (t *topics) CompactStatus(topic utils.TopicName) (utils.LongRunningProcessStatus, error) {
-	endpoint := t.client.endpoint(t.basePath, topic.GetRestPath(), "compaction")
+	endpoint := t.pulsar.endpoint(t.basePath, topic.GetRestPath(), "compaction")
 	var status utils.LongRunningProcessStatus
-	err := t.client.get(endpoint, &status)
+	err := t.pulsar.Client.Get(endpoint, &status)
 	return status, err
 }