prefix topics and consumer groups

Author: muir

broadcast.go

@@ -177,14 +177,15 @@ func (lib *Library[ID, TX, DB]) getBroadcastReader(ctx context.Context, broadcas
 
 	// We get a reader. Once we check that there is exactly one reader, we know we have exclusive use
 	// of the group because any other thread will see more than one reader
-	reader, _, readerConfig, err = lib.getReader(ctx, broadcastConsumerGroup, broadcastTopics(generic.Keys(lib.broadcast.topics)), true, resetPosition)
+	reader, _, readerConfig, err = lib.getReader(ctx, broadcastConsumerGroup, lib.addPrefixes(broadcastTopics(generic.Keys(lib.broadcast.topics))), true, resetPosition)
 	if err != nil {
 		// context was cancelled
 		return nil, nil, errors.WithStack(err)
 	}
 
 	b := backoffPolicy.Start(ctx)
 
+	prefixedBroadcastConsumerGroup := lib.addPrefix(string(broadcastConsumerGroup))
 FreshClient:
 	for {
 		// We need a new client. Either the group was deleted and thus the broker may have changed,
@@ -209,7 +210,7 @@ FreshClient:
 		ctxWithTimeout, cancelCtx = context.WithTimeout(ctx, describeTimeout)
 		describeResponse, err := client.DescribeGroups(ctxWithTimeout, &kafka.DescribeGroupsRequest{
 			Addr:     client.Addr,
-			GroupIDs: []string{string(broadcastConsumerGroup)},
+			GroupIDs: []string{prefixedBroadcastConsumerGroup},
 		})
 		cancelCtx()
 		if err != nil {
@@ -223,7 +224,7 @@ FreshClient:
 			return nil, nil, errors.Errorf("could not describe broadcast group: %w", err)
 		}
 		for _, resp := range describeResponse.Groups {
-			if resp.GroupID != string(broadcastConsumerGroup) {
+			if resp.GroupID != prefixedBroadcastConsumerGroup {
 				return nil, nil, errors.Errorf("got a description for a group (%s) that wasn't what was asked for (%s)", resp.GroupID, broadcastConsumerGroup)
 			}
 			switch {
@@ -271,7 +272,7 @@ func (lib *Library[ID, TX, DB]) deleteBroadcastConsumerGroup(ctx context.Context
 			lib.tracer.Logf("[events] deleting consumer group %s", broadcastConsumerGroup)
 			resp, err := client.DeleteGroups(ctxWithTimeout, &kafka.DeleteGroupsRequest{
 				Addr:     client.Addr,
-				GroupIDs: []string{string(broadcastConsumerGroup)},
+				GroupIDs: []string{lib.addPrefix(string(broadcastConsumerGroup))},
 			})
 			cancelCtx()
 			switch {
@@ -476,22 +477,22 @@ func (a *lockIDAllocator) grow() {
 	a.max = n
 }
 
-func broadcastTopics(topics []string) []string {
-	if !generic.SliceContainsElement(topics, heartbeatTopic.Topic()) {
+func broadcastTopics(unprefixedTopics []string) []string {
+	if !generic.SliceContainsElement(unprefixedTopics, heartbeatTopic.Topic()) {
 		// we don't actually need a handler, just need to subscribe to the heartbeat topic
-		topics = append(generic.CopySlice(topics), heartbeatTopic.Topic())
+		unprefixedTopics = append(generic.CopySlice(unprefixedTopics), heartbeatTopic.Topic())
 	}
-	return topics
+	return unprefixedTopics
 }
 
 // getReader only returns once the reader is actually connected working. This is determined
 // by calling Stats()
 // The only error that getReader returns is from context cancellation
-func (lib *Library[ID, TX, DB]) getReader(ctx context.Context, consumerGroup consumerGroupName, topics []string, isBroadcast bool, resetPosition bool) (*kafka.Reader, *kafka.ReaderStats, *kafka.ReaderConfig, error) {
+func (lib *Library[ID, TX, DB]) getReader(ctx context.Context, consumerGroup consumerGroupName, prefixedTopics []string, isBroadcast bool, resetPosition bool) (*kafka.Reader, *kafka.ReaderStats, *kafka.ReaderConfig, error) {
 	readerConfig := kafka.ReaderConfig{
 		Brokers:                lib.brokers,
-		GroupID:                consumerGroup.String(),
-		GroupTopics:            topics,
+		GroupID:                lib.addPrefix(consumerGroup.String()),
+		GroupTopics:            prefixedTopics,
 		MaxBytes:               maxBytes,
 		Dialer:                 lib.dialer(),
 		StartOffset:            kafka.FirstOffset,

connection.go

@@ -42,7 +42,7 @@ const (
 	broadcastStartupMaxWait             = 15 * time.Minute // only checked on group allocation failure
 	nonBroadcastReaderIdleTimeout       = 5 * time.Minute
 	broadcastReaderIdleTimeout          = baseBroadcastHeartbeat * 3 // This cannot be < 1s: Kafka takes a while to deliver events after reader startup or generation change
-	maxConsumerGroupNameLength          = 35
+	maxConsumerGroupNameLength          = 55                         // actual limit is 249 characters, but we need to subtract this from the topic name lenght limit so we use less
 	dialTimeout                         = time.Minute * 2
 	deleteTimeout                       = time.Minute * 4
 	describeTimeout                     = time.Minute * 4
@@ -80,7 +80,7 @@ const (
 )
 
 var (
-	maxTopicNameLength      = 255 - maxConsumerGroupNameLength - len(deadLetterTopicPostfix) - 1
+	maxTopicNameLength      = 249 - maxConsumerGroupNameLength - len(deadLetterTopicPostfix) - 1 // 249 is actual limit
 	LegalTopicNames         = regexp.MustCompile(fmt.Sprintf(`^[-._a-zA-Z0-9]{1,%d}$`, maxTopicNameLength))
 	legalConsumerGroupNames = regexp.MustCompile(fmt.Sprintf(`^[-._a-zA-Z0-9]{1,%d}$`, maxConsumerGroupNameLength))
 )
@@ -102,8 +102,8 @@ type LibraryNoDB struct {
 	broadcast                 *group
 	startTime                 time.Time
 	ready                     atomic.Int32
-	topicConfig               map[string]kafka.TopicConfig
-	topicsWork                pwork.Work[string, topicsWhy]
+	topicConfig               map[string]kafka.TopicConfig  // un-prefixed
+	topicsWork                pwork.Work[string, topicsWhy] // un-prefixed in APIs
 	topicListingStarted       sync.Once
 	topicsHaveBeenListed      chan struct{}
 	mustRegisterTopics        bool
@@ -124,6 +124,7 @@ type LibraryNoDB struct {
 	instanceID                int32
 	lazyProduce               bool
 	skipNotifier              bool
+	prefix                    string // prefixes all topics and consumer groups
 
 	// lock must be held when....
 	//
@@ -157,13 +158,13 @@ const (
 )
 
 type group struct {
-	topics  map[string]*topicHandlers
-	maxIdle time.Duration // reset reader when idle for this long
+	topics  map[string]*topicHandlers // unprefixed topic -> handlers
+	maxIdle time.Duration             // reset reader when idle for this long
 }
 
 type topicHandlers struct {
-	handlerNames []string // so that iteration is deterministic
-	handlers     map[string]*registeredHandler
+	handlerNames []string                      // so that iteration is deterministic
+	handlers     map[string]*registeredHandler // handlerName -> handler
 }
 
 type registeredHandler struct {
@@ -197,7 +198,7 @@ type handlerSuccess struct {
 type HandlerOpt func(*registeredHandler, *LibraryNoDB)
 
 type canHandle interface {
-	GetTopic() string
+	GetTopic() string // unprefixed
 	Handle(ctx context.Context, handlerInfo eventmodels.HandlerInfo, message []*kafka.Message) []error
 	Batch() bool
 }
@@ -262,6 +263,17 @@ func (lib *Library[ID, TX, DB]) SetBroadcastConsumerMaxLocks(max uint32) {
 	lib.broadcastConsumerMaxLock = max
 }
 
+// SetPrefix sets a string prefix used for all topics and all consumer groups. Keep this
+// short since max topic length is :
+//
+//	249 - 55 (maxConsumerGroupLength) - len("-dead-letter") - len(prefix) - 1
+func (lib *Library[ID, TX, DB]) SetPrefix(prefix string) {
+	lib.lock.Lock()
+	defer lib.lock.Unlock()
+	lib.mustNotBeRunning("attempt configure event library that is already processing")
+	lib.prefix = prefix
+}
+
 // DoNotLockBroadcastConsumerNumbers must be used before starting the library. If called,
 // no database locks will be taken to reserve broadcast consumer numbers. There is a
 // tradeoff: this saves a database connection that would otherwise sit around holding
@@ -670,7 +682,7 @@ func (lib *Library[ID, TX, DB]) getConsumerGroupCoordinator(ctx context.Context,
 	}
 	resp, err := controller.FindCoordinator(ctx, &kafka.FindCoordinatorRequest{
 		Addr:    controller.Addr,
-		Key:     consumerGroup.String(),
+		Key:     lib.addPrefix(consumerGroup.String()),
 		KeyType: kafka.CoordinatorKeyTypeConsumer,
 	})
 	if err != nil {
@@ -743,3 +755,34 @@ func NewConsumerGroup(name string) ConsumerGroupName {
 
 func (n consumerGroupName) String() string          { return string(n) }
 func (n consumerGroupName) name() consumerGroupName { return n }
+
+func (lib *LibraryNoDB) validateTopic(unprefixedTopic string) error {
+	if len(unprefixedTopic)+len(lib.prefix) > maxTopicNameLength {
+		return errors.Errorf("topic name (%s) is too long", unprefixedTopic)
+	}
+	if !LegalTopicNames.MatchString(unprefixedTopic) {
+		return errors.Errorf("topic name (%s) is invalid", unprefixedTopic)
+	}
+	return nil
+}
+
+// RemovePrefix removes a prefix from a topic or consumer group that was added
+// because the library had SetPrefix called previously.
+func (lib *LibraryNoDB) RemovePrefix(topicOrConsumerGroup string) string {
+	if lib.prefix == "" {
+		return topicOrConsumerGroup
+	}
+	return strings.TrimPrefix(topicOrConsumerGroup, lib.prefix)
+}
+
+func (lib *LibraryNoDB) addPrefix(topicOrConsumerGroup string) string {
+	return lib.prefix + topicOrConsumerGroup
+}
+
+func (lib *LibraryNoDB) addPrefixes(topicsOrConsumerGroups []string) []string {
+	p := make([]string, len(topicsOrConsumerGroups))
+	for i, unprefixed := range topicsOrConsumerGroups {
+		p[i] = lib.prefix + unprefixed
+	}
+	return p
+}

connection_test.go

@@ -11,5 +11,5 @@ func TestConstants(t *testing.T) {
 	t.Log("make sure nobody changes contants in ways that would break behavior")
 	assert.Greater(t, broadcastReaderIdleTimeout, time.Second)
 	assert.Less(t, broadcastHeartbeatRandom, 0.8)
-	assert.Less(t, maxConsumerGroupNameLength, 36)
+	assert.Less(t, maxConsumerGroupNameLength, 56)
 }