Repository: apache/pulsar-client-go
Pull request #743 (2022-03-09 15:34:07)

Add TableView support
Fixes https://github.com/apache/pulsar/issues/13929

### Motivation

This adds TableView support to the Golang client so it can have feature parity with the Java client.

See https://github.com/apache/pulsar/issues/12356 and https://github.com/apache/pulsar/pull/12838 for context.

### Modifications

Add `TableView` and `Client.CreateTableView` and so users can track a keyed topic as a map.

### Verifying this change

- [ ] Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

- `go test -timeout 5s -run ^TestTableView$ github.com/apache/pulsar-client-go/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? GoDocs


ZiyaoWei: A few things worth noting:

- This is stacked on top of https://github.com/apache/pulsar-client-go/pull/743/;
- The Java implementation actually return views of the underlying map in, e.g., `Keys()`, which I think might be problematic since then the users could inadvertently change the entries in the `TableView`. Maybe we should change it to returning a copy of the underlying values (or maybe I am missing something)?
- The internal table is a `map` which aren't thread safe - I'll try changing it to `sync.Map` which should make the code less verbose, although since Golang doesn't have generics internally we have to do some type assertions. Alternatively, if we are okay with changing the accessors to return copies we can do the same and stick with a `map`.
- Since (again) Golang doesn't support generics yet there's no good way of knowing what the types of the values would be, so I am making the API similar to `GetSchemaValue(v interface{})` by letting the users pass the `reflect.Type` of the values. This feels a bit ugly but I can't think of a better alternative, suggestions are most welcome!

Patch:
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/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..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 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)))
+	}
+}