*

internal/config/config.go

@@ -15,7 +15,6 @@ type smtpConfig struct {
 
 type kafkaConfig struct {
 	Port int    `envconfig:"KAFKA_PORT"`
-	Partition int    `envconfig:"KAFKA_PARTITION"`
 	Host string `envconfig:"KAFKA_HOST"`
 }
 

internal/postman.go

@@ -3,10 +3,12 @@ package postman
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"log"
 	"net"
 	"strconv"
 	"test3k/authPostman/internal/config"
+	"time"
 
 	// smtp "test3k/authPostman/internal/smtp"
 
@@ -31,10 +33,11 @@ func NewServer(ctx context.Context, config *config.Config, topic string) *AuthPo
 		kafkaReader: kafka.NewReader(kafka.ReaderConfig{
 			Topic:   topic,
 			Brokers: []string{net.JoinHostPort(config.Kafka.Host, strconv.Itoa(config.Kafka.Port))},
-			// GroupID:   fmt.Sprintf("consumer-group-%d", config.Kafka.Partition),
+			GroupID: fmt.Sprintf("consumer-group-%s", topic),
-			Partition: config.Kafka.Partition,
+			// Partition:      0,
 			MinBytes:       10e3, // 10KB
 			MaxBytes:       10e6, // 10MB
+			ReadBackoffMax: time.Millisecond * 100,
 		}),
 	}
 }
@@ -45,7 +48,7 @@ func (s *AuthPostmanServer) GracefulStop() error {
 
 func (s *AuthPostmanServer) ReadMessage(offset int64) error {
 	// ...
-	s.kafkaReader.SetOffset(offset)
+	// s.kafkaReader.SetOffset(offset)
 
 	//
 	for {