Repository: apache/pulsar-client-go (go)

Issue:
Panic occurred when a producer associated with a KeyBasedBatchBuilder.
Function internalFlushCurrentBatch, [ this line of code](https://github.com/apache/pulsar-client-go/blob/bb2fa811a0f5e61bc978b627b89d00a17f68b436/pulsar/producer_partition.go#L470), should not be invoked if the batch builder associated with the producer is a multi batches container.

Potential Fix:
Using internalFlushCurrentBatches instead, if the associated batch builder is a multi batches container.

```
		if p.batchBuilder.IsMultiBatches() {
			p.internalFlushCurrentBatches()
		} else {
			p.internalFlushCurrentBatch()
		}
```   


Pull request:
[Issue 749]Fix panic caused by flushing current batch with an incorrect internal function.
…rrentBatches if the associated batch builder contains multi batches(KeyBasedBatchBuilder).

Doing transparent forwarding could centralized the branching logic at only one place.

<--
### Contribution Checklist
  
  - Name the pull request in the form "[Issue XYZ][component] Title of the pull request", where *XYZ* should be replaced by the actual issue number.
    Skip *Issue XYZ* if there is no associated github issue for this pull request.
    Skip *component* if you are unsure about which is the best component. E.g. `[docs] Fix typo in produce method`.

  - Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
  
  - Each pull request should address only one issue, not mix up code from multiple issues.
  
  - Each commit in the pull request has a meaningful commit message

  - Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
-->

Fixes #749 

### Motivation
Eliminating multiple identical patches, which branch to the correct internal function by checking the type of associated batch builder.

### Modifications

Calling internalFlushCurrentBatch will be forwarded to internalFlushCurrentBatches if the associated batch builder contains multiple batches(e.g. KeyBasedBatchBuilder). 

### Verifying this change

- [x] Make sure that the change passes the CI checks.

This change is already covered by existing integration tests.

### Does this pull request potentially affect one of the following parts:

None

### Documentation

Not required

Patch:
diff --git a/perf/pulsar-perf-go.go b/perf/pulsar-perf-go.go
index 488a74f..aff0b73 100644
--- a/perf/pulsar-perf-go.go
+++ b/perf/pulsar-perf-go.go
@@ -41,16 +41,18 @@ var flagDebug bool
 var PrometheusPort int
 
 type ClientArgs struct {
-	ServiceURL       string
-	TokenFile        string
-	TLSTrustCertFile string
+	ServiceURL              string
+	TokenFile               string
+	TLSTrustCertFile        string
+	MaxConnectionsPerBroker int
 }
 
 var clientArgs ClientArgs
 
 func NewClient() (pulsar.Client, error) {
 	clientOpts := pulsar.ClientOptions{
-		URL: clientArgs.ServiceURL,
+		URL:                     clientArgs.ServiceURL,
+		MaxConnectionsPerBroker: clientArgs.MaxConnectionsPerBroker,
 	}
 
 	if clientArgs.TokenFile != "" {
@@ -97,6 +99,8 @@ func main() {
 		"pulsar://localhost:6650", "The Pulsar service URL")
 	flags.StringVar(&clientArgs.TokenFile, "token-file", "", "file path to the Pulsar JWT file")
 	flags.StringVar(&clientArgs.TLSTrustCertFile, "trust-cert-file", "", "file path to the trusted certificate file")
+	flags.IntVarP(&clientArgs.MaxConnectionsPerBroker, "max-connections", "c", 1,
+		"Max connections to open to broker. Defaults to 1.")
 
 	rootCmd.AddCommand(newProducerCommand())
 	rootCmd.AddCommand(newConsumerCommand())
diff --git a/pulsar/client.go b/pulsar/client.go
index 8ff152a..f4642c6 100644
--- a/pulsar/client.go
+++ b/pulsar/client.go
@@ -141,6 +141,10 @@ type Client interface {
 	// This method will block until the reader is created successfully.
 	CreateReader(ReaderOptions) (Reader, error)
 
+	// CreateTableView creates a table view instance.
+	// This method will block until the table view is created successfully.
+	CreateTableView(TableViewOptions) (TableView, error)
+
 	// TopicPartitions Fetches the list of partitions for a given topic
 	//
 	// If the topic is partitioned, this will return a list of partition names.
diff --git a/pulsar/client_impl.go b/pulsar/client_impl.go
index 5682927..09b72af 100644
--- a/pulsar/client_impl.go
+++ b/pulsar/client_impl.go
@@ -176,6 +176,15 @@ func (c *client) CreateReader(options ReaderOptions) (Reader, error) {
 	return reader, nil
 }
 
+func (c *client) CreateTableView(options TableViewOptions) (TableView, error) {
+	tableView, err := newTableView(c, options)
+	if err != nil {
+		return nil, err
+	}
+	c.handlers.Add(tableView)
+	return tableView, nil
+}
+
 func (c *client) TopicPartitions(topic string) ([]string, error) {
 	topicName, err := internal.ParseTopicName(topic)
 	if err != nil {
diff --git a/pulsar/consumer.go b/pulsar/consumer.go
index a71a2d4..c67509b 100644
--- a/pulsar/consumer.go
+++ b/pulsar/consumer.go
@@ -241,8 +241,8 @@ type Consumer interface {
 	// Note: this operation can only be done on non-partitioned topics. For these, one can rather perform the seek() on
 	// the individual partitions.
 	//
-	// @param timestamp
-	//            the message publish time where to reposition the subscription
+	// @param time
+	//            the message publish time when to reposition the subscription
 	//
 	SeekByTime(time time.Time) error
 
diff --git a/pulsar/dlq_router.go b/pulsar/dlq_router.go
index c858946..b28600f 100644
--- a/pulsar/dlq_router.go
+++ b/pulsar/dlq_router.go
@@ -89,8 +89,7 @@ func (r *dlqRouter) run() {
 		select {
 		case cm := <-r.messageCh:
 			r.log.WithField("msgID", cm.ID()).Debug("Got message for DLQ")
-			producer := r.getProducer()
-
+			producer := r.getProducer(cm.Consumer.(*consumer).options.Schema)
 			msg := cm.Message.(*message)
 			msgID := msg.ID()
 			producer.SendAsync(context.Background(), &ProducerMessage{
@@ -127,7 +126,7 @@ func (r *dlqRouter) close() {
 	}
 }
 
-func (r *dlqRouter) getProducer() Producer {
+func (r *dlqRouter) getProducer(schema Schema) Producer {
 	if r.producer != nil {
 		// Producer was already initialized
 		return r.producer
@@ -140,6 +139,7 @@ func (r *dlqRouter) getProducer() Producer {
 			Topic:                   r.policy.DeadLetterTopic,
 			CompressionType:         LZ4,
 			BatchingMaxPublishDelay: 100 * time.Millisecond,
+			Schema:                  schema,
 		})
 
 		if err != nil {
diff --git a/pulsar/error.go b/pulsar/error.go
index f433bfc..ead5cf9 100644
--- a/pulsar/error.go
+++ b/pulsar/error.go
@@ -101,6 +101,8 @@ const (
 	SeekFailed
 	// ProducerClosed means producer already been closed
 	ProducerClosed
+	// SchemaFailure means the payload could not be encoded using the Schema
+	SchemaFailure
 )
 
 // Error implement error interface, composed of two parts: msg and result.
@@ -205,6 +207,8 @@ func getResultStr(r Result) string {
 		return "SeekFailed"
 	case ProducerClosed:
 		return "ProducerClosed"
+	case SchemaFailure:
+		return "SchemaFailure"
 	default:
 		return fmt.Sprintf("Result(%d)", r)
 	}
diff --git a/pulsar/internal/connection.go b/pulsar/internal/connection.go
index a025abf..11c9d49 100644
--- a/pulsar/internal/connection.go
+++ b/pulsar/internal/connection.go
@@ -53,7 +53,8 @@ type TLSOptions struct {
 }
 
 var (
-	errConnectionClosed = errors.New("connection closed")
+	errConnectionClosed       = errors.New("connection closed")
+	errUnableRegisterListener = errors.New("unable register listener when con closed")
 )
 
 // ConnectionListener is a user of a connection (eg. a producer or
@@ -72,7 +73,7 @@ type Connection interface {
 	SendRequest(requestID uint64, req *pb.BaseCommand, callback func(*pb.BaseCommand, error))
 	SendRequestNoWait(req *pb.BaseCommand) error
 	WriteData(data Buffer)
-	RegisterListener(id uint64, listener ConnectionListener)
+	RegisterListener(id uint64, listener ConnectionListener) error
 	UnregisterListener(id uint64)
 	AddConsumeHandler(id uint64, handler ConsumerHandler)
 	DeleteConsumeHandler(id uint64)
@@ -522,10 +523,12 @@ func (c *connection) internalReceivedCommand(cmd *pb.BaseCommand, headersAndPayl
 		c.handleResponse(cmd.ProducerSuccess.GetRequestId(), cmd)
 
 	case pb.BaseCommand_PARTITIONED_METADATA_RESPONSE:
+		c.checkServerError(cmd.PartitionMetadataResponse.Error)
 		c.handleResponse(cmd.PartitionMetadataResponse.GetRequestId(), cmd)
 
 	case pb.BaseCommand_LOOKUP_RESPONSE:
 		lookupResult := cmd.LookupTopicResponse
+		c.checkServerError(lookupResult.Error)
 		c.handleResponse(lookupResult.GetRequestId(), cmd)
 
 	case pb.BaseCommand_CONSUMER_STATS_RESPONSE:
@@ -574,6 +577,16 @@ func (c *connection) internalReceivedCommand(cmd *pb.BaseCommand, headersAndPayl
 	}
 }
 
+func (c *connection) checkServerError(err *pb.ServerError) {
+	if err == nil {
+		return
+	}
+
+	if *err == pb.ServerError_ServiceNotReady {
+		c.Close()
+	}
+}
+
 func (c *connection) Write(data Buffer) {
 	c.writeRequestsCh <- data
 }
@@ -835,17 +848,18 @@ func (c *connection) handleCloseProducer(closeProducer *pb.CommandCloseProducer)
 	}
 }
 
-func (c *connection) RegisterListener(id uint64, listener ConnectionListener) {
+func (c *connection) RegisterListener(id uint64, listener ConnectionListener) error {
 	// do not add if connection is closed
 	if c.closed() {
 		c.log.Warnf("Connection closed unable register listener id=%+v", id)
-		return
+		return errUnableRegisterListener
 	}
 
 	c.listenersLock.Lock()
 	defer c.listenersLock.Unlock()
 
 	c.listeners[id] = listener
+	return nil
 }
 
 func (c *connection) UnregisterListener(id uint64) {
diff --git a/pulsar/producer_partition.go b/pulsar/producer_partition.go
index d031e9a..87d1d69 100644
--- a/pulsar/producer_partition.go
+++ b/pulsar/producer_partition.go
@@ -262,11 +262,14 @@ func (p *partitionProducer) grabCnx() error {
 		p.sequenceIDGenerator = &nextSequenceID
 	}
 	p._setConn(res.Cnx)
-	p._getConn().RegisterListener(p.producerID, p)
+	err = p._getConn().RegisterListener(p.producerID, p)
+	if err != nil {
+		return err
+	}
 	p.log.WithFields(log.Fields{
 		"cnx":   res.Cnx.ID(),
 		"epoch": atomic.LoadUint64(&p.epoch),
-	}).Debug("Connected producer")
+	}).Info("Connected producer")
 
 	pendingItems := p.pendingQueue.ReadableSlice()
 	viewSize := len(pendingItems)
@@ -383,11 +386,7 @@ func (p *partitionProducer) runEventsLoop() {
 				return
 			}
 		case <-p.batchFlushTicker.C:
-			if p.batchBuilder.IsMultiBatches() {
-				p.internalFlushCurrentBatches()
-			} else {
-				p.internalFlushCurrentBatch()
-			}
+			p.internalFlushCurrentBatch()
 		}
 	}
 }
@@ -405,18 +404,22 @@ func (p *partitionProducer) internalSend(request *sendRequest) {
 
 	msg := request.msg
 
+	// read payload from message
 	payload := msg.Payload
-	var schemaPayload []byte
+
 	var err error
-	if p.options.Schema != nil {
+
+	// payload and schema are mutually exclusive
+	// try to get payload from schema value only if payload is not set
+	if payload == nil && p.options.Schema != nil {
+		var schemaPayload []byte
 		schemaPayload, err = p.options.Schema.Encode(msg.Value)
 		if err != nil {
+			p.publishSemaphore.Release()
+			request.callback(nil, request.msg, newError(SchemaFailure, err.Error()))
 			p.log.WithError(err).Errorf("Schema encode message failed %s", msg.Value)
 			return
 		}
-	}
-
-	if payload == nil {
 		payload = schemaPayload
 	}
 
@@ -478,11 +481,8 @@ func (p *partitionProducer) internalSend(request *sendRequest) {
 		msg.ReplicationClusters, deliverAt)
 	if !added {
 		// The current batch is full.. flush it and retry
-		if p.batchBuilder.IsMultiBatches() {
-			p.internalFlushCurrentBatches()
-		} else {
-			p.internalFlushCurrentBatch()
-		}
+
+		p.internalFlushCurrentBatch()
 
 		// after flushing try again to add the current payload
 		if ok := p.batchBuilder.Add(smm, p.sequenceIDGenerator, payload, request,
@@ -497,11 +497,9 @@ func (p *partitionProducer) internalSend(request *sendRequest) {
 	}
 
 	if !sendAsBatch || request.flushImmediately {
-		if p.batchBuilder.IsMultiBatches() {
-			p.internalFlushCurrentBatches()
-		} else {
-			p.internalFlushCurrentBatch()
-		}
+
+		p.internalFlushCurrentBatch()
+
 	}
 }
 
@@ -515,6 +513,11 @@ type pendingItem struct {
 }
 
 func (p *partitionProducer) internalFlushCurrentBatch() {
+	if p.batchBuilder.IsMultiBatches() {
+		p.internalFlushCurrentBatches()
+		return
+	}
+
 	batchData, sequenceID, callbacks, err := p.batchBuilder.Flush()
 	if batchData == nil {
 		return
@@ -676,11 +679,8 @@ func (p *partitionProducer) internalFlushCurrentBatches() {
 }
 
 func (p *partitionProducer) internalFlush(fr *flushRequest) {
-	if p.batchBuilder.IsMultiBatches() {
-		p.internalFlushCurrentBatches()
-	} else {
-		p.internalFlushCurrentBatch()
-	}
+
+	p.internalFlushCurrentBatch()
 
 	pi, ok := p.pendingQueue.PeekLast().(*pendingItem)
 	if !ok {
@@ -780,22 +780,22 @@ func (p *partitionProducer) ReceivedSendReceipt(response *pb.CommandSendReceipt)
 
 	if !ok {
 		// if we receive a receipt although the pending queue is empty, the state of the broker and the producer differs.
-		// At that point, it is better to close the connection to the broker to reconnect to a broker hopping it solves
-		// the state discrepancy.
-		p.log.Warnf("Received ack for %v although the pending queue is empty, closing connection", response.GetMessageId())
-		p._getConn().Close()
+		p.log.Warnf("Got ack %v for timed out msg", response.GetMessageId())
 		return
 	}
 
 	if pi.sequenceID < response.GetSequenceId() {
-		// if we receive a receipt that is not the one expected, the state of the broker and the producer differs.
-		// At that point, it is better to close the connection to the broker to reconnect to a broker hopping it solves
-		// the state discrepancy.
+		// Ignoring the ack since it's referring to a message that has already timed out.
+		p.log.Warnf("Received ack for %v on sequenceId %v - expected: %v, closing connection", response.GetMessageId(),
+			response.GetSequenceId(), pi.sequenceID)
+		return
+	} else if pi.sequenceID > response.GetSequenceId() {
+		// Force connection closing so that messages can be re-transmitted in a new connection
 		p.log.Warnf("Received ack for %v on sequenceId %v - expected: %v, closing connection", response.GetMessageId(),
 			response.GetSequenceId(), pi.sequenceID)
 		p._getConn().Close()
 		return
-	} else if pi.sequenceID == response.GetSequenceId() {
+	} else {
 		// The ack was indeed for the expected item in the queue, we can remove it and trigger the callback
 		p.pendingQueue.Poll()
 
diff --git a/pulsar/reader.go b/pulsar/reader.go
index f1cb575..3539037 100644
--- a/pulsar/reader.go
+++ b/pulsar/reader.go
@@ -68,6 +68,10 @@ type ReaderOptions struct {
 	// SubscriptionRolePrefix sets the subscription role prefix. The default prefix is "reader".
 	SubscriptionRolePrefix string
 
+	// SubscriptionName sets the subscription name.
+	// If subscriptionRolePrefix is set at the same time, this configuration will prevail
+	SubscriptionName string
+
 	// ReadCompacted, if enabled, the reader will read messages from the compacted topic rather than reading the
 	// full message backlog of the topic. This means that, if the topic has been compacted, the reader will only
 	// see the latest value for each key in the topic, up until the point in the topic message backlog that has
diff --git a/pulsar/reader_impl.go b/pulsar/reader_impl.go
index 596884a..dd552b6 100644
--- a/pulsar/reader_impl.go
+++ b/pulsar/reader_impl.go
@@ -66,11 +66,14 @@ func newReader(client *client, options ReaderOptions) (Reader, error) {
 		}
 	}
 
-	subscriptionName := options.SubscriptionRolePrefix
+	subscriptionName := options.SubscriptionName
 	if subscriptionName == "" {
-		subscriptionName = "reader"
+		subscriptionName = options.SubscriptionRolePrefix
+		if subscriptionName == "" {
+			subscriptionName = "reader"
+		}
+		subscriptionName += "-" + generateRandomName()
 	}
-	subscriptionName += "-" + generateRandomName()
 
 	receiverQueueSize := options.ReceiverQueueSize
 	if receiverQueueSize <= 0 {
diff --git a/pulsar/table_view.go b/pulsar/table_view.go
new file mode 100644
index 0000000..e566bf0
--- /dev/null
+++ b/pulsar/table_view.go
@@ -0,0 +1,78 @@
+// 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 (
+	"reflect"
+	"time"
+
+	"github.com/apache/pulsar-client-go/pulsar/log"
+)
+
+// TableViewOptions contains the options for creating a TableView
+type TableViewOptions struct {
+	// Topic specifies the topic this table view will subscribe on.
+	// This argument is required when constructing the table view.
+	Topic string
+
+	// Set the interval of updating partitions. Default to 1 minute.
+	AutoUpdatePartitionsInterval time.Duration
+
+	// Schema represents the schema implementation.
+	Schema Schema
+
+	// SchemaValueType represents the type of values for the given schema.
+	SchemaValueType reflect.Type
+
+	// Configure the logger used by the TableView.
+	// By default, a wrapped logrus.StandardLogger will be used, namely,
+	// log.NewLoggerWithLogrus(logrus.StandardLogger())
+	Logger log.Logger
+}
+
+// TableView provides a key-value map view of a compacted topic. Messages without keys will be ignored.
+type TableView interface {
+	// Size returns the number of key-value mappings in the TableView.
+	Size() int
+
+	// IsEmpty returns true if this TableView contains no key-value mappings.
+	IsEmpty() bool
+
+	// ContainsKey returns true if this TableView contains a mapping for the specified key.
+	ContainsKey(key string) bool
+
+	// Get returns the value to which the specified key is mapped, or nil if this map contains no mapping for the key.
+	Get(key string) interface{}
+
+	// Entries returns a map view of the mappings contained in this TableView.
+	Entries() map[string]interface{}
+
+	// Keys returns a slice of the keys contained in this TableView.
+	Keys() []string
+
+	// ForEach performs the give action for each entry in this map until all entries have been processed or the action
+	// returns an error.
+	ForEach(func(string, interface{}) error) error
+
+	// ForEachAndListen performs the give action for each entry in this map until all entries have been processed or
+	// the action returns an error.
+	ForEachAndListen(func(string, interface{}) error) error
+
+	// Close closes the table view and releases resources allocated.
+	Close()
+}
diff --git a/pulsar/table_view_impl.go b/pulsar/table_view_impl.go
new file mode 100644
index 0000000..56b9fcb
--- /dev/null
+++ b/pulsar/table_view_impl.go
@@ -0,0 +1,271 @@
+// 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 (
+	"context"
+	"errors"
+	"fmt"
+	"reflect"
+	"sync"
+	"time"
+
+	"github.com/apache/pulsar-client-go/pulsar/log"
+	"github.com/sirupsen/logrus"
+)
+
+type cancelReader struct {
+	reader     Reader
+	cancelFunc context.CancelFunc
+}
+
+type TableViewImpl struct {
+	client  *client
+	options TableViewOptions
+
+	dataMu sync.Mutex
+	data   map[string]interface{}
+
+	readersMu    sync.Mutex
+	cancelRaders map[string]cancelReader
+
+	listenersMu sync.Mutex
+	listeners   []func(string, interface{}) error
+
+	logger   log.Logger
+	closed   bool
+	closedCh chan struct{}
+}
+
+func newTableView(client *client, options TableViewOptions) (TableView, error) {
+	if options.Topic == "" {
+		return nil, newError(TopicNotFound, "topic is required")
+	}
+
+	if options.Schema != nil && options.SchemaValueType == nil {
+		return nil, newError(InvalidConfiguration, "SchemaValueType is required when Schema is present")
+	}
+
+	var logger log.Logger
+	if options.Logger != nil {
+		logger = options.Logger
+	} else {
+		logger = log.NewLoggerWithLogrus(logrus.StandardLogger())
+	}
+
+	if options.AutoUpdatePartitionsInterval == 0 {
+		options.AutoUpdatePartitionsInterval = time.Minute
+	}
+
+	tv := TableViewImpl{
+		client:       client,
+		options:      options,
+		data:         make(map[string]interface{}),
+		cancelRaders: make(map[string]cancelReader),
+		logger:       logger,
+		closedCh:     make(chan struct{}),
+	}
+
+	// Do an initial round of partition update check to make sure we can populate the partition readers
+	if err := tv.partitionUpdateCheck(); err != nil {
+		return nil, err
+	}
+	go tv.periodicPartitionUpdateCheck()
+
+	return &tv, nil
+}
+
+func (tv *TableViewImpl) partitionUpdateCheck() error {
+	partitionsArray, err := tv.client.TopicPartitions(tv.options.Topic)
+	if err != nil {
+		return fmt.Errorf("tv.client.TopicPartitions(%s) failed: %w", tv.options.Topic, err)
+	}
+
+	partitions := make(map[string]bool, len(partitionsArray))
+	for _, partition := range partitionsArray {
+		partitions[partition] = true
+	}
+
+	tv.readersMu.Lock()
+	defer tv.readersMu.Unlock()
+
+	for partition, cancelReader := range tv.cancelRaders {
+		if _, ok := partitions[partition]; !ok {
+			cancelReader.cancelFunc()
+			cancelReader.reader.Close()
+			delete(tv.cancelRaders, partition)
+		}
+	}
+
+	for partition := range partitions {
+		if _, ok := tv.cancelRaders[partition]; !ok {
+			reader, err := newReader(tv.client, ReaderOptions{
+				Topic:          partition,
+				StartMessageID: EarliestMessageID(),
+				ReadCompacted:  true,
+				// TODO: Pooling?
+				Schema: tv.options.Schema,
+			})
+			if err != nil {
+				return fmt.Errorf("create new reader failed for %s: %w", partition, err)
+			}
+			for reader.HasNext() {
+				msg, err := reader.Next(context.Background())
+				if err != nil {
+					tv.logger.Errorf("read next message failed for %s: %w", partition, err)
+				}
+				tv.handleMessage(msg)
+			}
+			ctx, cancelFunc := context.WithCancel(context.Background())
+			tv.cancelRaders[partition] = cancelReader{
+				reader:     reader,
+				cancelFunc: cancelFunc,
+			}
+			go tv.watchReaderForNewMessages(ctx, reader)
+		}
+	}
+
+	return nil
+}
+
+func (tv *TableViewImpl) periodicPartitionUpdateCheck() {
+	for {
+		if err := tv.partitionUpdateCheck(); err != nil {
+			tv.logger.Errorf("failed to check for changes in number of partitions: %w", err)
+		}
+		select {
+		case <-tv.closedCh:
+			// If the TableViewImpl has been closed, stop checking for partition updates
+			return
+		case <-time.After(tv.options.AutoUpdatePartitionsInterval):
+			continue
+		}
+	}
+}
+
+func (tv *TableViewImpl) Size() int {
+	tv.dataMu.Lock()
+	defer tv.dataMu.Unlock()
+	return len(tv.data)
+}
+
+func (tv *TableViewImpl) IsEmpty() bool {
+	tv.dataMu.Lock()
+	defer tv.dataMu.Unlock()
+	return tv.Size() == 0
+}
+
+func (tv *TableViewImpl) ContainsKey(key string) bool {
+	tv.dataMu.Lock()
+	defer tv.dataMu.Unlock()
+	_, ok := tv.data[key]
+	return ok
+}
+
+func (tv *TableViewImpl) Get(key string) interface{} {
+	tv.dataMu.Lock()
+	defer tv.dataMu.Unlock()
+	return tv.data[key]
+}
+
+func (tv *TableViewImpl) Entries() map[string]interface{} {
+	tv.dataMu.Lock()
+	defer tv.dataMu.Unlock()
+	data := make(map[string]interface{}, len(tv.data))
+	for k, v := range tv.data {
+		data[k] = v
+	}
+	return tv.data
+}
+
+func (tv *TableViewImpl) Keys() []string {
+	tv.dataMu.Lock()
+	defer tv.dataMu.Unlock()
+	keys := make([]string, len(tv.data))
+	i := 0
+	for k := range tv.data {
+		keys[i] = k
+		i++
+	}
+	return keys
+}
+
+func (tv *TableViewImpl) ForEach(action func(string, interface{}) error) error {
+	tv.dataMu.Lock()
+	defer tv.dataMu.Unlock()
+	for k, v := range tv.data {
+		if err := action(k, v); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (tv *TableViewImpl) ForEachAndListen(action func(string, interface{}) error) error {
+	tv.listenersMu.Lock()
+	defer tv.listenersMu.Unlock()
+
+	if err := tv.ForEach(action); err != nil {
+		return err
+	}
+
+	tv.listeners = append(tv.listeners, action)
+	return nil
+}
+
+func (tv *TableViewImpl) Close() {
+	tv.readersMu.Lock()
+	defer tv.readersMu.Unlock()
+
+	if !tv.closed {
+		tv.closed = true
+		for _, cancelReader := range tv.cancelRaders {
+			cancelReader.reader.Close()
+		}
+		close(tv.closedCh)
+	}
+}
+
+func (tv *TableViewImpl) handleMessage(msg Message) {
+	tv.dataMu.Lock()
+	defer tv.dataMu.Unlock()
+
+	payload := reflect.Indirect(reflect.New(tv.options.SchemaValueType)).Interface()
+	if err := msg.GetSchemaValue(&payload); err != nil {
+		tv.logger.Errorf("msg.GetSchemaValue() failed with %w; msg is %v", msg, err)
+	}
+	tv.data[msg.Key()] = payload
+	for _, listener := range tv.listeners {
+		if err := listener(msg.Key(), payload); err != nil {
+			tv.logger.Errorf("table view listener failed for %v: %w", msg, err)
+		}
+	}
+}
+
+func (tv *TableViewImpl) watchReaderForNewMessages(ctx context.Context, reader Reader) {
+	for {
+		msg, err := reader.Next(ctx)
+		if err != nil {
+			tv.logger.Errorf("read next message failed for %s: %w", reader.Topic(), err)
+		}
+		if errors.Is(err, context.Canceled) {
+			return
+		}
+		tv.handleMessage(msg)
+	}
+}


Test patch:
diff --git a/pulsar/client_impl_test.go b/pulsar/client_impl_test.go
index e9e94af..2d83c99 100644
--- a/pulsar/client_impl_test.go
+++ b/pulsar/client_impl_test.go
@@ -348,7 +348,9 @@ func TestTopicPartitions(t *testing.T) {
 	defer client.Close()
 
 	// Create topic with 5 partitions
-	err = httpPut("admin/v2/persistent/public/default/TestGetTopicPartitions/partitions", 5)
+	topicAdminURL := "admin/v2/persistent/public/default/TestGetTopicPartitions/partitions"
+	err = httpPut(topicAdminURL, 5)
+	defer httpDelete(topicAdminURL)
 	assert.Nil(t, err)
 
 	partitionedTopic := "persistent://public/default/TestGetTopicPartitions"
@@ -697,7 +699,9 @@ func TestHTTPTopicPartitions(t *testing.T) {
 	defer client.Close()
 
 	// Create topic with 5 partitions
-	err = httpPut("admin/v2/persistent/public/default/TestHTTPTopicPartitions/partitions", 5)
+	topicAdminURL := "admin/v2/persistent/public/default/TestHTTPTopicPartitions/partitions"
+	err = httpPut(topicAdminURL, 5)
+	defer httpDelete(topicAdminURL)
 	assert.Nil(t, err)
 
 	partitionedTopic := "persistent://public/default/TestHTTPTopicPartitions"
diff --git a/pulsar/producer_test.go b/pulsar/producer_test.go
index 124c828..541c1fe 100644
--- a/pulsar/producer_test.go
+++ b/pulsar/producer_test.go
@@ -509,7 +509,9 @@ func TestRoundRobinRouterPartitionedProducer(t *testing.T) {
 
 func TestMessageRouter(t *testing.T) {
 	// Create topic with 5 partitions
-	err := httpPut("admin/v2/persistent/public/default/my-partitioned-topic/partitions", 5)
+	topicAdminURL := "admin/v2/persistent/public/default/my-partitioned-topic/partitions"
+	err := httpPut(topicAdminURL, 5)
+	defer httpDelete(topicAdminURL)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -922,6 +924,50 @@ func TestMaxMessageSize(t *testing.T) {
 	}
 }
 
+func TestFailedSchemaEncode(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: lookupURL,
+	})
+
+	assert.Nil(t, err)
+	defer client.Close()
+
+	topic := newTopicName()
+	ctx := context.Background()
+
+	// create producer
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic:  topic,
+		Schema: NewAvroSchema("{\"type\":\"string\"}", nil),
+	})
+
+	assert.Nil(t, err)
+	defer producer.Close()
+
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		// producer should send return an error as message is Int64, but schema is String
+		mid, err := producer.Send(ctx, &ProducerMessage{
+			Value: int64(1),
+		})
+		assert.NotNil(t, err)
+		assert.Nil(t, mid)
+		wg.Done()
+	}()
+
+	wg.Add(1)
+	// producer should send return an error as message is Int64, but schema is String
+	producer.SendAsync(ctx, &ProducerMessage{
+		Value: int64(1),
+	}, func(messageID MessageID, producerMessage *ProducerMessage, err error) {
+		assert.NotNil(t, err)
+		assert.Nil(t, messageID)
+		wg.Done()
+	})
+	wg.Wait()
+}
+
 func TestSendTimeout(t *testing.T) {
 	quotaURL := adminURL + "/admin/v2/namespaces/public/default/backlogQuota"
 	quotaFmt := `{"limit": "%d", "policy": "producer_request_hold"}`
diff --git a/pulsar/reader_test.go b/pulsar/reader_test.go
index aa12078..3c85871 100644
--- a/pulsar/reader_test.go
+++ b/pulsar/reader_test.go
@@ -24,6 +24,7 @@ import (
 	"time"
 
 	"github.com/apache/pulsar-client-go/pulsar/crypto"
+	"github.com/google/uuid"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -48,6 +49,27 @@ func TestReaderConfigErrors(t *testing.T) {
 	assert.NotNil(t, err)
 }
 
+func TestReaderConfigSubscribeName(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: lookupURL,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer client.Close()
+
+	consumer, err := client.CreateReader(ReaderOptions{
+		StartMessageID:   EarliestMessageID(),
+		Topic:            uuid.New().String(),
+		SubscriptionName: uuid.New().String(),
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer consumer.Close()
+	assert.NotNil(t, consumer)
+}
+
 func TestReader(t *testing.T) {
 	client, err := NewClient(ClientOptions{
 		URL: lookupURL,
diff --git a/pulsar/table_view_test.go b/pulsar/table_view_test.go
new file mode 100644
index 0000000..829187b
--- /dev/null
+++ b/pulsar/table_view_test.go
@@ -0,0 +1,80 @@
+// 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 (
+	"context"
+	"fmt"
+	"reflect"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestTableView(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: lookupURL,
+	})
+
+	assert.NoError(t, err)
+	defer client.Close()
+
+	topic := newTopicName()
+	schema := NewStringSchema(nil)
+
+	// create producer
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic:  topic,
+		Schema: schema,
+	})
+	assert.NoError(t, err)
+	defer producer.Close()
+
+	numMsg := 10
+	valuePrefix := "hello pulsar: "
+	for i := 0; i < numMsg; i++ {
+		key := fmt.Sprintf("%d", i)
+		t.Log(key)
+		_, err = producer.Send(context.Background(), &ProducerMessage{
+			Key:   key,
+			Value: fmt.Sprintf(valuePrefix + key),
+		})
+		assert.NoError(t, err)
+	}
+
+	// create table view
+	v := ""
+	tv, err := client.CreateTableView(TableViewOptions{
+		Topic:           topic,
+		Schema:          schema,
+		SchemaValueType: reflect.TypeOf(&v),
+	})
+	assert.NoError(t, err)
+	defer tv.Close()
+
+	// Wait until tv receives all messages
+	for tv.Size() < 10 {
+		time.Sleep(time.Second * 1)
+		t.Logf("TableView number of elements: %d", tv.Size())
+	}
+
+	for k, v := range tv.Entries() {
+		assert.Equal(t, valuePrefix+k, *(v.(*string)))
+	}
+}