Repository: apache/pulsar-client-go
Pull request #723 (2022-02-14 09:19:36)

Dlq producer on topic with schema
- create dlq producer with schema from consumer 
- address issue #726 
- follow java client to create dlq producer [here](https://github.com/apache/pulsar/blob/master/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java#L1984)


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 2e43ddc..2e0c97d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -23,6 +23,63 @@
 
 All notable changes to this project will be documented in this file.
 
+[0.8.1] 2022-03-08
+
+## What's Changed
+* Upgrade beefsack/go-rate by @shubham1172 in https://github.com/apache/pulsar-client-go/pull/735
+* Upgrade prometheus client to 1.11.1 by @nicoloboschi in https://github.com/apache/pulsar-client-go/pull/738
+
+## New Contributors
+@shubham1172 made their first contribution in https://github.com/apache/pulsar-client-go/pull/735
+@nicoloboschi made their first contribution in https://github.com/apache/pulsar-client-go/pull/738
+
+[0.8.0] 2022-02-16
+
+## What's Changed
+* Update release docs with missing information by @cckellogg in https://github.com/apache/pulsar-client-go/pull/656
+* Update change log for 0.7.0 release by @cckellogg in https://github.com/apache/pulsar-client-go/pull/655
+* Update version to 0.7.0 by @cckellogg in https://github.com/apache/pulsar-client-go/pull/654
+* fix issue 650,different handle sequence value by @baomingyu in https://github.com/apache/pulsar-client-go/pull/651
+* Support nack backoff policy for SDK by @wolfstudy in https://github.com/apache/pulsar-client-go/pull/660
+* Remove unused dependency in `oauth2` module by @reugn in https://github.com/apache/pulsar-client-go/pull/661
+* [Issue 662] Fix race in connection.go waitUntilReady() by @bschofield in https://github.com/apache/pulsar-client-go/pull/663
+* Update dependencies by @reugn in https://github.com/apache/pulsar-client-go/pull/665
+* [Issue 652] Quick fixes to the documentation for the main building blocks of the library by @reugn in https://github.com/apache/pulsar-client-go/pull/667
+* Add subscription properties for ConsumerOptions by @wolfstudy in https://github.com/apache/pulsar-client-go/pull/671
+* Add new bug-resistant build constraints by @reugn in https://github.com/apache/pulsar-client-go/pull/670
+* Handle the parameters parsing error in NewProvider by @reugn in https://github.com/apache/pulsar-client-go/pull/673
+* Update email template of release docs by @izumo27 in https://github.com/apache/pulsar-client-go/pull/674
+* Add properties filed for batch container by @wolfstudy in https://github.com/apache/pulsar-client-go/pull/683
+* [Issue 513] Correct apparent logic error in batchContainer's hasSpace() method by @bschofield in https://github.com/apache/pulsar-client-go/pull/678
+* Upgrade DataDog/zstd to v1.5.0 by @dferstay in https://github.com/apache/pulsar-client-go/pull/690
+* fix:add order key to message by @leizhiyuan in https://github.com/apache/pulsar-client-go/pull/688
+* Set default go version to 1.13 in CI related files by @reugn in https://github.com/apache/pulsar-client-go/pull/692
+* [Partition Consumer] Provide lock-free access to compression providers by @dferstay in https://github.com/apache/pulsar-client-go/pull/689
+* Use a separate gorutine to handle the logic of reconnect by @wolfstudy in https://github.com/apache/pulsar-client-go/pull/691
+* [DefaultRouter] add a parallel bench test by @dferstay in https://github.com/apache/pulsar-client-go/pull/693
+* Revert "Use a separate gorutine to handle the logic of reconnect" by @wolfstudy in https://github.com/apache/pulsar-client-go/pull/700
+* Fix data race while accessing connection in partitionProducer by @wolfstudy in https://github.com/apache/pulsar-client-go/pull/701
+* Fix stuck when reconnect broker by @wolfstudy in https://github.com/apache/pulsar-client-go/pull/703
+* Fix slice bounds out of range for readSingleMessage by @wolfstudy in https://github.com/apache/pulsar-client-go/pull/709
+* Encryption failure test case fix by @GPrabhudas in https://github.com/apache/pulsar-client-go/pull/708
+* [DefaultRouter] fix unnecessary system clock reads due to races accessing router state by @dferstay in https://github.com/apache/pulsar-client-go/pull/694
+* Fix negative WaitGroup counter issue by @wolfstudy in https://github.com/apache/pulsar-client-go/pull/712
+* [issue 675] oauth2 use golang-jwt address CVE-2020-26160 by @zzzming in https://github.com/apache/pulsar-client-go/pull/713
+* readme: add note about how to build and test by @pgier in https://github.com/apache/pulsar-client-go/pull/714
+* Bump oauth2 package version to the latest in master by @iorvd in https://github.com/apache/pulsar-client-go/pull/715
+* Fix closed connection leak by @billowqiu in https://github.com/apache/pulsar-client-go/pull/716
+* [Bugfix] producer runEventsLoop for reconnect early exit by @billowqiu in https://github.com/apache/pulsar-client-go/pull/721
+* [issue 679][oauth2] Fix macos compiler warnings by @pgier in https://github.com/apache/pulsar-client-go/pull/719
+* [optimize] Stop batchFlushTicker when Disable batching by @Shoothzj in https://github.com/apache/pulsar-client-go/pull/720
+* Markdown error fix by @Shoothzj in https://github.com/apache/pulsar-client-go/pull/722
+
+## New Contributors
+* @bschofield made their first contribution in https://github.com/apache/pulsar-client-go/pull/663
+* @izumo27 made their first contribution in https://github.com/apache/pulsar-client-go/pull/674
+* @pgier made their first contribution in https://github.com/apache/pulsar-client-go/pull/714
+* @iorvd made their first contribution in https://github.com/apache/pulsar-client-go/pull/715
+* @billowqiu made their first contribution in https://github.com/apache/pulsar-client-go/pull/716
+
 [0.7.0] 2021-10-31
 
 ## Feature
diff --git a/Dockerfile b/Dockerfile
index 401a246..e12cc80 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -17,8 +17,8 @@
 # under the License.
 #
 
-ARG GO_VERSION=golang:1.13
-FROM apachepulsar/pulsar:latest as pulsar
+ARG GO_VERSION=golang:1.15
+FROM apachepulsar/pulsar:2.8.2 as pulsar
 FROM $GO_VERSION as go
 
 RUN apt-get update && apt-get install -y openjdk-11-jre-headless ca-certificates
diff --git a/VERSION b/VERSION
index 3129f0c..c12365b 100644
--- a/VERSION
+++ b/VERSION
@@ -1,3 +1,3 @@
 // This version number refers to the currently released version number
 // Please fix the version when release.
-v0.7.0
+v0.8.0
diff --git a/go.mod b/go.mod
index c788991..ec6d810 100644
--- a/go.mod
+++ b/go.mod
@@ -6,18 +6,18 @@ require (
 	github.com/AthenZ/athenz v1.10.39
 	github.com/DataDog/zstd v1.5.0
 	github.com/apache/pulsar-client-go/oauth2 v0.0.0-20220120090717-25e59572242e
-	github.com/beefsack/go-rate v0.0.0-20180408011153-efa7637bb9b6
+	github.com/beefsack/go-rate v0.0.0-20220214233405-116f4ca011a0
 	github.com/bmizerany/perks v0.0.0-20141205001514-d9a9656a3a4b
 	github.com/davecgh/go-spew v1.1.1
 	github.com/gogo/protobuf v1.3.2
 	github.com/golang/protobuf v1.5.2
 	github.com/google/uuid v1.1.2
-	github.com/klauspost/compress v1.10.8
+	github.com/klauspost/compress v1.14.4
 	github.com/linkedin/goavro/v2 v2.9.8
 	github.com/opentracing/opentracing-go v1.2.0
 	github.com/pierrec/lz4 v2.0.5+incompatible
 	github.com/pkg/errors v0.9.1
-	github.com/prometheus/client_golang v1.11.0
+	github.com/prometheus/client_golang v1.11.1
 	github.com/sirupsen/logrus v1.6.0
 	github.com/spaolacci/murmur3 v1.1.0
 	github.com/spf13/cobra v1.2.1
diff --git a/go.sum b/go.sum
index 0e0fe1c..4eb4433 100644
--- a/go.sum
+++ b/go.sum
@@ -60,8 +60,8 @@ github.com/armon/circbuf v0.0.0-20150827004946-bbbad097214e/go.mod h1:3U/XgcO3hC
 github.com/armon/go-metrics v0.0.0-20180917152333-f0300d1749da/go.mod h1:Q73ZrmVTwzkszR9V5SSuryQ31EELlFMUz1kKyl939pY=
 github.com/armon/go-radix v0.0.0-20180808171621-7fddfc383310/go.mod h1:ufUuZ+zHj4x4TnLV4JWEpy2hxWSpsRywHrMgIH9cCH8=
 github.com/aws/aws-sdk-go v1.32.6/go.mod h1:5zCpMtNQVjRREroY7sYe8lOMRSxkhG6MZveU8YkpAk0=
-github.com/beefsack/go-rate v0.0.0-20180408011153-efa7637bb9b6 h1:KXlsf+qt/X5ttPGEjR0tPH1xaWWoKBEg9Q1THAj2h3I=
-github.com/beefsack/go-rate v0.0.0-20180408011153-efa7637bb9b6/go.mod h1:6YNgTHLutezwnBvyneBbwvB8C82y3dcoOj5EQJIdGXA=
+github.com/beefsack/go-rate v0.0.0-20220214233405-116f4ca011a0 h1:0b2vaepXIfMsG++IsjHiI2p4bxALD1Y2nQKGMR5zDQM=
+github.com/beefsack/go-rate v0.0.0-20220214233405-116f4ca011a0/go.mod h1:6YNgTHLutezwnBvyneBbwvB8C82y3dcoOj5EQJIdGXA=
 github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q=
 github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+CedLV8=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
@@ -234,8 +234,8 @@ github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7V
 github.com/julienschmidt/httprouter v1.3.0/go.mod h1:JR6WtHb+2LUe8TCKY3cZOxFyyO8IZAc4RVcycCCAKdM=
 github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
-github.com/klauspost/compress v1.10.8 h1:eLeJ3dr/Y9+XRfJT4l+8ZjmtB5RPJhucH2HeCV5+IZY=
-github.com/klauspost/compress v1.10.8/go.mod h1:aoV0uJVorq1K+umq18yTdKaF57EivdYsUV+/s2qKfXs=
+github.com/klauspost/compress v1.14.4 h1:eijASRJcobkVtSt81Olfh7JX43osYLwy5krOJo6YEu4=
+github.com/klauspost/compress v1.14.4/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk=
 github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/konsorten/go-windows-terminal-sequences v1.0.3 h1:CE8S1cTafDpPvMhIxNJKvHsGVBgn1xWYf1NbHQhywc8=
 github.com/konsorten/go-windows-terminal-sequences v1.0.3/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
@@ -298,8 +298,8 @@ github.com/posener/complete v1.1.1/go.mod h1:em0nMJCgc9GFtwrmVmEMR/ZL6WyhyjMBndr
 github.com/prometheus/client_golang v0.9.1/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXPKyh/dDVn+NZz0KFw=
 github.com/prometheus/client_golang v1.0.0/go.mod h1:db9x61etRT2tGnBNRi70OPL5FsnadC4Ky3P0J6CfImo=
 github.com/prometheus/client_golang v1.7.1/go.mod h1:PY5Wy2awLA44sXw4AOSfFBetzPP4j5+D6mVACh+pe2M=
-github.com/prometheus/client_golang v1.11.0 h1:HNkLOAEQMIDv/K+04rukrLx6ch7msSRwf3/SASFAGtQ=
-github.com/prometheus/client_golang v1.11.0/go.mod h1:Z6t4BnS23TR94PD6BsDNk8yVqroYurpAkEiz0P2BEV0=
+github.com/prometheus/client_golang v1.11.1 h1:+4eQaD7vAZ6DsfsxB15hbE0odUjGI5ARs9yskGu1v4s=
+github.com/prometheus/client_golang v1.11.1/go.mod h1:Z6t4BnS23TR94PD6BsDNk8yVqroYurpAkEiz0P2BEV0=
 github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910/go.mod h1:MbSGuTsp3dbXC40dX6PRTWyKYBIrTGTE9sqQNg2J8bo=
 github.com/prometheus/client_model v0.0.0-20190129233127-fd36f4220a90/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
 github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
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_partition.go b/pulsar/consumer_partition.go
index 04a39c5..b06474d 100644
--- a/pulsar/consumer_partition.go
+++ b/pulsar/consumer_partition.go
@@ -498,12 +498,17 @@ func (pc *partitionConsumer) MessageReceived(response *pb.CommandMessage, header
 	pbMsgID := response.GetMessageId()
 
 	reader := internal.NewMessageReader(headersAndPayload)
+	brokerMetadata, err := reader.ReadBrokerMetadata()
+	if err != nil {
+		// todo optimize use more appropriate error codes
+		pc.discardCorruptedMessage(pbMsgID, pb.CommandAck_BatchDeSerializeError)
+		return err
+	}
 	msgMeta, err := reader.ReadMessageMetadata()
 	if err != nil {
 		pc.discardCorruptedMessage(pbMsgID, pb.CommandAck_ChecksumMismatch)
 		return err
 	}
-
 	decryptedPayload, err := pc.decryptor.Decrypt(headersAndPayload.ReadableSlice(), pbMsgID, msgMeta)
 	// error decrypting the payload
 	if err != nil {
@@ -597,7 +602,18 @@ func (pc *partitionConsumer) MessageReceived(response *pb.CommandMessage, header
 			pc.AckID(msgID)
 			continue
 		}
-
+		var messageIndex *uint64
+		var brokerPublishTime *time.Time
+		if brokerMetadata != nil {
+			if brokerMetadata.Index != nil {
+				aux := brokerMetadata.GetIndex() - uint64(numMsgs) + uint64(i) + 1
+				messageIndex = &aux
+			}
+			if brokerMetadata.BrokerTimestamp != nil {
+				aux := timeFromUnixTimestampMillis(*brokerMetadata.BrokerTimestamp)
+				brokerPublishTime = &aux
+			}
+		}
 		// set the consumer so we know how to ack the message id
 		msgID.consumer = pc
 		var msg *message
@@ -616,6 +632,8 @@ func (pc *partitionConsumer) MessageReceived(response *pb.CommandMessage, header
 				replicatedFrom:      msgMeta.GetReplicatedFrom(),
 				redeliveryCount:     response.GetRedeliveryCount(),
 				orderingKey:         string(smm.OrderingKey),
+				index:               messageIndex,
+				brokerPublishTime:   brokerPublishTime,
 			}
 		} else {
 			msg = &message{
@@ -631,6 +649,8 @@ func (pc *partitionConsumer) MessageReceived(response *pb.CommandMessage, header
 				replicationClusters: msgMeta.GetReplicateTo(),
 				replicatedFrom:      msgMeta.GetReplicatedFrom(),
 				redeliveryCount:     response.GetRedeliveryCount(),
+				index:               messageIndex,
+				brokerPublishTime:   brokerPublishTime,
 			}
 		}
 
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/impl_message.go b/pulsar/impl_message.go
index a9809aa..3216676 100644
--- a/pulsar/impl_message.go
+++ b/pulsar/impl_message.go
@@ -241,6 +241,8 @@ type message struct {
 	redeliveryCount     uint32
 	schema              Schema
 	encryptionContext   *EncryptionContext
+	index               *uint64
+	brokerPublishTime   *time.Time
 }
 
 func (msg *message) Topic() string {
@@ -299,6 +301,14 @@ func (msg *message) GetEncryptionContext() *EncryptionContext {
 	return msg.encryptionContext
 }
 
+func (msg *message) Index() *uint64 {
+	return msg.index
+}
+
+func (msg *message) BrokerPublishTime() *time.Time {
+	return msg.brokerPublishTime
+}
+
 func newAckTracker(size int) *ackTracker {
 	var batchIDs *big.Int
 	if size <= 64 {
diff --git a/pulsar/internal/buffer.go b/pulsar/internal/buffer.go
index f3b8fe6..b3e23fb 100644
--- a/pulsar/internal/buffer.go
+++ b/pulsar/internal/buffer.go
@@ -38,6 +38,8 @@ type Buffer interface {
 
 	Read(size uint32) []byte
 
+	Skip(size uint32)
+
 	Get(readerIndex uint32, size uint32) []byte
 
 	ReadableSlice() []byte
@@ -122,6 +124,10 @@ func (b *buffer) Read(size uint32) []byte {
 	return res
 }
 
+func (b *buffer) Skip(size uint32) {
+	b.readerIdx += size
+}
+
 func (b *buffer) Get(readerIdx uint32, size uint32) []byte {
 	return b.data[readerIdx : readerIdx+size]
 }
diff --git a/pulsar/internal/commands.go b/pulsar/internal/commands.go
index b91c0b6..7fd1885 100644
--- a/pulsar/internal/commands.go
+++ b/pulsar/internal/commands.go
@@ -18,6 +18,7 @@
 package internal
 
 import (
+	"encoding/binary"
 	"errors"
 	"fmt"
 
@@ -34,8 +35,9 @@ const (
 	// MessageFramePadding is for metadata and other frame headers
 	MessageFramePadding = 10 * 1024
 	// MaxFrameSize limit the maximum size that pulsar allows for messages to be sent.
-	MaxFrameSize        = MaxMessageSize + MessageFramePadding
-	magicCrc32c  uint16 = 0x0e01
+	MaxFrameSize                    = MaxMessageSize + MessageFramePadding
+	magicCrc32c              uint16 = 0x0e01
+	magicBrokerEntryMetadata uint16 = 0x0e02
 )
 
 // ErrCorruptedMessage is the error returned by ReadMessageData when it has detected corrupted data.
@@ -119,6 +121,20 @@ func (r *MessageReader) ReadMessageMetadata() (*pb.MessageMetadata, error) {
 	return &meta, nil
 }
 
+func (r *MessageReader) ReadBrokerMetadata() (*pb.BrokerEntryMetadata, error) {
+	magicNumber := binary.BigEndian.Uint16(r.buffer.Get(r.buffer.ReaderIndex(), 2))
+	if magicNumber != magicBrokerEntryMetadata {
+		return nil, nil
+	}
+	r.buffer.Skip(2)
+	size := r.buffer.ReadUint32()
+	var brokerEntryMetadata pb.BrokerEntryMetadata
+	if err := proto.Unmarshal(r.buffer.Read(size), &brokerEntryMetadata); err != nil {
+		return nil, err
+	}
+	return &brokerEntryMetadata, nil
+}
+
 func (r *MessageReader) ReadMessage() (*pb.SingleMessageMetadata, []byte, error) {
 	if r.buffer.ReadableBytes() == 0 && r.buffer.Capacity() > 0 {
 		return nil, nil, ErrEOM
diff --git a/pulsar/internal/connection.go b/pulsar/internal/connection.go
index 1e2009c..6055252 100644
--- a/pulsar/internal/connection.go
+++ b/pulsar/internal/connection.go
@@ -43,7 +43,7 @@ const (
 	PulsarVersion       = "0.1"
 	ClientVersionString = "Pulsar Go " + PulsarVersion
 
-	PulsarProtocolVersion = int32(pb.ProtocolVersion_v13)
+	PulsarProtocolVersion = int32(pb.ProtocolVersion_v18)
 )
 
 type TLSOptions struct {
@@ -292,7 +292,8 @@ func (c *connection) doHandshake() bool {
 		AuthMethodName:  proto.String(c.auth.Name()),
 		AuthData:        authData,
 		FeatureFlags: &pb.FeatureFlags{
-			SupportsAuthRefresh: proto.Bool(true),
+			SupportsAuthRefresh:         proto.Bool(true),
+			SupportsBrokerEntryMetadata: proto.Bool(true),
 		},
 	}
 
@@ -521,10 +522,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:
@@ -573,6 +576,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
 }
diff --git a/pulsar/message.go b/pulsar/message.go
index 3779caf..b88f158 100644
--- a/pulsar/message.go
+++ b/pulsar/message.go
@@ -121,6 +121,14 @@ type Message interface {
 	// GetEncryptionContext returns the ecryption context of the message.
 	// It will be used by the application to parse the undecrypted message.
 	GetEncryptionContext() *EncryptionContext
+
+	// Index returns index from broker entry metadata,
+	// or empty if the feature is not enabled in the broker.
+	Index() *uint64
+
+	// BrokerPublishTime returns broker publish time from broker entry metadata,
+	// or empty if the feature is not enabled in the broker.
+	BrokerPublishTime() *time.Time
 }
 
 // MessageID identifier for a particular message
diff --git a/pulsar/producer_partition.go b/pulsar/producer_partition.go
index 0e2e4c4..d37d60e 100644
--- a/pulsar/producer_partition.go
+++ b/pulsar/producer_partition.go
@@ -145,6 +145,7 @@ func newPartitionProducer(client *client, topic string, options *ProducerOptions
 	}
 	err := p.grabCnx()
 	if err != nil {
+		p.batchFlushTicker.Stop()
 		logger.WithError(err).Error("Failed to create producer at newPartitionProducer")
 		return nil, err
 	}
@@ -404,18 +405,20 @@ 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.log.WithError(err).Errorf("Schema encode message failed %s", msg.Value)
 			return
 		}
-	}
-
-	if payload == nil {
 		payload = schemaPayload
 	}
 
@@ -779,22 +782,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 c45b8ff..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
@@ -79,6 +83,9 @@ type ReaderOptions struct {
 
 	// Decryption represents the encryption related fields required by the reader to decrypt a message.
 	Decryption *MessageDecryptionInfo
+
+	// Schema represents the schema implementation.
+	Schema Schema
 }
 
 // Reader can be used to scan through all the messages currently available in a topic.
diff --git a/pulsar/reader_impl.go b/pulsar/reader_impl.go
index 0fed80c..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 {
@@ -102,6 +105,7 @@ func newReader(client *client, options ReaderOptions) (Reader, error) {
 		nackRedeliveryDelay:        defaultNackRedeliveryDelay,
 		replicateSubscriptionState: false,
 		decryption:                 options.Decryption,
+		schema:                     options.Schema,
 	}
 
 	reader := &reader{
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)
+	}
+}
diff --git a/stable.txt b/stable.txt
index eb53fa6..aff2506 100644
--- a/stable.txt
+++ b/stable.txt
@@ -1,3 +1,3 @@
 // This version number refers to the current stable version, generally is `VERSION - 1`.
 // Please fix the version when release.
-v0.7.0
+v0.8.0


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/internal/commands_test.go b/pulsar/internal/commands_test.go
index b43335a..c236e10 100644
--- a/pulsar/internal/commands_test.go
+++ b/pulsar/internal/commands_test.go
@@ -70,6 +70,19 @@ func TestReadMessageMetadata(t *testing.T) {
 	assert.Equal(t, 10, int(meta.GetNumMessagesInBatch()))
 }
 
+func TestReadBrokerEntryMetadata(t *testing.T) {
+	// read old style message (not batched)
+	reader := NewMessageReaderFromArray(brokerEntryMeta)
+	meta, err := reader.ReadBrokerMetadata()
+	if err != nil {
+		t.Fatal(err)
+	}
+	var expectedBrokerTimestamp uint64 = 1646983036054
+	assert.Equal(t, expectedBrokerTimestamp, *meta.BrokerTimestamp)
+	var expectedIndex uint64 = 5
+	assert.Equal(t, expectedIndex, *meta.Index)
+}
+
 func TestReadMessageOldFormat(t *testing.T) {
 	reader := NewMessageReaderFromArray(rawCompatSingleMessage)
 	_, err := reader.ReadMessageMetadata()
@@ -210,3 +223,8 @@ var rawBatchMessage10 = []byte{
 	0x28, 0x05, 0x40, 0x09, 0x68, 0x65, 0x6c, 0x6c,
 	0x6f,
 }
+
+var brokerEntryMeta = []byte{
+	0x0e, 0x02, 0x00, 0x00, 0x00, 0x09, 0x08, 0x96,
+	0xf9, 0xda, 0xbe, 0xf7, 0x2f, 0x10, 0x05,
+}
diff --git a/pulsar/internal/pulsartracing/message_carrier_util_test.go b/pulsar/internal/pulsartracing/message_carrier_util_test.go
index 677a7ff..7f25578 100644
--- a/pulsar/internal/pulsartracing/message_carrier_util_test.go
+++ b/pulsar/internal/pulsartracing/message_carrier_util_test.go
@@ -123,3 +123,11 @@ func (msg *mockConsumerMessage) ProducerName() string {
 func (msg *mockConsumerMessage) GetEncryptionContext() *pulsar.EncryptionContext {
 	return &pulsar.EncryptionContext{}
 }
+
+func (msg *mockConsumerMessage) Index() *uint64 {
+	return nil
+}
+
+func (msg *mockConsumerMessage) BrokerPublishTime() *time.Time {
+	return nil
+}
diff --git a/pulsar/negative_acks_tracker_test.go b/pulsar/negative_acks_tracker_test.go
index 51965ea..e47fb09 100644
--- a/pulsar/negative_acks_tracker_test.go
+++ b/pulsar/negative_acks_tracker_test.go
@@ -237,6 +237,14 @@ func (msg *mockMessage1) GetEncryptionContext() *EncryptionContext {
 	return &EncryptionContext{}
 }
 
+func (msg *mockMessage1) Index() *uint64 {
+	return nil
+}
+
+func (msg *mockMessage1) BrokerPublishTime() *time.Time {
+	return nil
+}
+
 type mockMessage2 struct {
 	properties map[string]string
 }
@@ -300,3 +308,11 @@ func (msg *mockMessage2) ProducerName() string {
 func (msg *mockMessage2) GetEncryptionContext() *EncryptionContext {
 	return &EncryptionContext{}
 }
+
+func (msg *mockMessage2) Index() *uint64 {
+	return nil
+}
+
+func (msg *mockMessage2) BrokerPublishTime() *time.Time {
+	return nil
+}
diff --git a/pulsar/producer_test.go b/pulsar/producer_test.go
index 124c828..3c526bb 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)
 	}
diff --git a/pulsar/reader_test.go b/pulsar/reader_test.go
index bdafea0..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,
@@ -710,3 +732,46 @@ func TestProducerReaderRSAEncryption(t *testing.T) {
 		assert.Equal(t, []byte(expectMsg), msg.Payload())
 	}
 }
+
+func TestReaderWithSchema(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: lookupURL,
+	})
+
+	assert.Nil(t, err)
+	defer client.Close()
+
+	topic := newTopicName()
+	schema := NewStringSchema(nil)
+
+	// create producer
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic:  topic,
+		Schema: schema,
+	})
+	assert.Nil(t, err)
+	defer producer.Close()
+
+	value := "hello pulsar"
+	_, err = producer.Send(context.Background(), &ProducerMessage{
+		Value: value,
+	})
+	assert.Nil(t, err)
+
+	// create reader
+	reader, err := client.CreateReader(ReaderOptions{
+		Topic:          topic,
+		StartMessageID: EarliestMessageID(),
+		Schema:         schema,
+	})
+	assert.Nil(t, err)
+	defer reader.Close()
+
+	msg, err := reader.Next(context.Background())
+	assert.NoError(t, err)
+
+	var res *string
+	err = msg.GetSchemaValue(&res)
+	assert.Nil(t, err)
+	assert.Equal(t, *res, value)
+}
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)))
+	}
+}