pubsub/ AWS MSK Support

[alok87]

Does pubsub support AWS MSK? (aws serveless kafka)

AWS MSK requires these special token providers https://github.com/aws/aws-msk-iam-sasl-signer-go

[vangent]

As far as I know, it should. The Go CDK constructors take a sarama.Config, and it looks like the required token providers are provided on that.

[vangent]

The default URL opener will probably not work, as it uses a MinimalConfig and there's no way to override anything from the URL. However, you can create your own URLOpener with the saram.Config set:

https://github.com/google/go-cloud/blob/master/pubsub/kafkapubsub/kafka.go#L152

OpenTopic is here:

https://github.com/google/go-cloud/blob/master/pubsub/kafkapubsub/kafka.go#L232

[alok87]

Did not get? will the following work or not!

• OpenTopics
• Subscription.Receive()

my code is doing below, wont it work?

package main

func main() {
        config := cfg.New(appEnv)

        config.Pub.Kafka.Sarama = kafkapubsub.MinimalConfig()
	config.Pub.Kafka.Sarama.Net.SASL.Enable = true
	config.Pub.Kafka.Sarama.Net.SASL.Mechanism = sarama.SASLTypeOAuth
	config.Pub.Kafka.Sarama.Net.SASL.TokenProvider = pubsub.NewMSKAccessTokenProvider("ap-south-1")
	log.Infof(ctx, "MSK Bootstrap server: %v", config.Pub.Kafka.Address)


        pubService, err := pub.New(
		ctx,
		config.Pub,
		[]string{"mytopic"},
	)

        pubService.Publish()
}

package pub

import (
	"context"
	"fmt"
	"sync"

	"github.com/IBM/sarama"
	gopubsub "gocloud.dev/pubsub"
	"gocloud.dev/pubsub/kafkapubsub"

	"github.com/myorg/myworld/pkg/log"
)

// Pub is a wrapper over gocloud pubsub
// to reduce code duplication across modules
type Pub struct {
	// openTopics keeps a list of open topics
	openTopics map[string]*gopubsub.Topic

	// config can be used later also
	config Config
}

// Config has the pubsub configuration
type Config struct {
	// Kafka configurations
	Kafka *KafkaConfig
}

// KafkaConfig keeps all kafka related config
type KafkaConfig struct {
	// Sarama kafka library configurations
	Sarama *sarama.Config
	// Address where kafka is present
	Address []string
}

// New creates a new instance of pubsub, opens subscriptions, and creates topics.
func New(
	ctx context.Context,
	config Config,
	openTopics []string,
) (*Pub, error) {
	if config.Kafka == nil {
		return nil, fmt.Errorf("only kafka supported currently")
	}

	openedTopics := make(map[string]*gopubsub.Topic)
	for _, topicName := range openTopics {
		topic, err := kafkapubsub.OpenTopic(
			config.Kafka.Address,
			config.Kafka.Sarama,
			topicName,
			nil,
		)
		if err != nil {
			return nil, fmt.Errorf(
				"could not create/establish conn with topic: %+v, err:%v",
				topicName,
				err,
			)
		}
		openedTopics[topicName] = topic
	}

	return &Pub{
		openTopics: openedTopics,
		config:     config,
	}, nil
}

// Close closes all open subscriptions and topics.
func (ps *Pub) Close(ctx context.Context, wg *sync.WaitGroup) {
	defer wg.Done()

	<-ctx.Done()
	for _, topic := range ps.openTopics {
		err := topic.Shutdown(ctx)
		if err != nil {
			log.Errorf(ctx, "error closing pubsub topic: %v", err)
		}
	}
}

// OpenTopic creates and registers a new topic.
func (ps *Pub) OpenTopic(
	ctx context.Context,
	topicName string,
) error {
	if _, exists := ps.openTopics[topicName]; exists {
		return fmt.Errorf("topic: %v already registered", topicName)
	}

	topic, err := kafkapubsub.OpenTopic(
		ps.config.Kafka.Address,
		ps.config.Kafka.Sarama,
		topicName,
		nil,
	)
	if err != nil {
		return fmt.Errorf(
			"could not create/establish conn with topic: %+v, err:%v",
			topicName,
			err,
		)
	}
	ps.openTopics[topicName] = topic

	return nil
}

// Publish writes to the topic.
func (ps *Pub) Publish(
	ctx context.Context,
	topicName string,
	message []byte,
) error {
	topic, ok := ps.openTopics[topicName]
	if !ok || topic == nil {
		return fmt.Errorf("topic: %v unregistered", topic)
	}

	return topic.Send(
		ctx,
		&gopubsub.Message{
			Body: message,
		},
	)
}

I am currently testing only above using go-cloud, only opening topics and publishing.

[vangent]

As far as I know if you set up the config.Sarama correctly for AWS MSK, it should work with Go CDK. If you are having problems I suggest you try it using the native AWS MSK to make sure your config is correct.

[alok87]

Getting this kafka: client has run out of available brokers to talk to)

I debugged also

It is failing at: https://github.com/google/go-cloud/blob/master/pubsub/kafkapubsub/kafka.go#L247-L250

brokers configuration is correct, i tested with

sh-4.2# ./kafka-topics.sh --bootstrap-server $BS --command-config client.properties --list

[alok87]

Working, TLS config was issue, thanks gocloud works with MSK 👍🏽