Repository: apache/pulsar-client-go
Pull request #741 (2022-03-07 21:57:31)

Add schema support to Reader
Master Issue: #344

### Motivation

#368 adds support for schemas in consumer and producer, but didn't add the corresponding option to reader. This fixes that.

### Modifications

Add `schema` field to `ReaderOptions` and pass the value to the consumer so messages can be read with a schema.

### Verifying this change

- [ ] Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

`go test -run ^TestReaderWithSchema$ ./pulsar`

### Does this pull request potentially affect one of the following parts:

  - The public API: yes

### Documentation

  - Does this pull request introduce a new feature? yes
  - If yes, how is the feature documented? GoDoc


Shoothzj: @ZiyaoWei #748 is fixed, you can rebase the code for retrigger the ci

Patch:
diff --git a/Dockerfile b/Dockerfile
index a21a219..e12cc80 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -17,7 +17,7 @@
 # under the License.
 #
 
-ARG GO_VERSION=golang:1.13
+ARG GO_VERSION=golang:1.15
 FROM apachepulsar/pulsar:2.8.2 as pulsar
 FROM $GO_VERSION as go
 
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/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..a025abf 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),
 		},
 	}
 
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/reader.go b/pulsar/reader.go
index c45b8ff..f1cb575 100644
--- a/pulsar/reader.go
+++ b/pulsar/reader.go
@@ -79,6 +79,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..596884a 100644
--- a/pulsar/reader_impl.go
+++ b/pulsar/reader_impl.go
@@ -102,6 +102,7 @@ func newReader(client *client, options ReaderOptions) (Reader, error) {
 		nackRedeliveryDelay:        defaultNackRedeliveryDelay,
 		replicateSubscriptionState: false,
 		decryption:                 options.Decryption,
+		schema:                     options.Schema,
 	}
 
 	reader := &reader{


Test patch:
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/reader_test.go b/pulsar/reader_test.go
index bdafea0..aa12078 100644
--- a/pulsar/reader_test.go
+++ b/pulsar/reader_test.go
@@ -710,3 +710,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)
+}