sleep?

backend/runner/pubsub/consumer.go

@@ -89,6 +89,10 @@ func (c *consumer) Begin(ctx context.Context) error {
 	if err != nil {
 		return fmt.Errorf("failed to create consumer group for subscription %s: %w", c.verb.Name, err)
 	}
+	c.checkOffsets(ctx)
+
+	time.Sleep(1 * time.Second)
+
 	go c.watchErrors(ctx, group)
 	go c.subscribe(ctx, group)
 	return nil
@@ -226,3 +230,38 @@ func (c *consumer) call(ctx context.Context, body []byte, partition, offset int)
 	observability.Calls.Request(ctx, req.Verb, start, optional.None[string]())
 	return nil
 }
+
+func (c *consumer) checkOffsets(ctx context.Context) {
+	groupId := kafkaConsumerGroupID(c.moduleName, c.verb)
+	logger := log.FromContext(ctx).Scope(groupId + ".offsets")
+	client, err := sarama.NewClient(c.verb.Runtime.Subscription.KafkaBrokers, sarama.NewConfig())
+	if err != nil {
+		logger.Errorf(err, "failed to create client")
+	}
+	admin, err := sarama.NewClusterAdminFromClient(client)
+	if err != nil {
+		logger.Errorf(err, "failed to create cluster admin")
+		return
+	}
+	groups, err := admin.DescribeConsumerGroups([]string{groupId})
+	if err != nil {
+		logger.Errorf(err, "failed to describe consumer groups")
+		return
+	}
+	if len(groups) == 0 {
+		logger.Infof("no consumer group found")
+		return
+	}
+	resp, err := admin.ListConsumerGroupOffsets(groupId, nil)
+	if err != nil {
+		logger.Errorf(err, "failed to list consumer group offsets")
+		return
+	}
+	outStrs := []string{}
+	for topic, partitions := range resp.Blocks {
+		for partition, block := range partitions {
+			outStrs = append(outStrs, fmt.Sprintf("%s:%d:%d", topic, partition, block.Offset))
+		}
+	}
+	logger.Infof("offsets:\n%s", strings.Join(outStrs, "\n"))
+}