consumer_setup.go

func readerKafkaSetup(
	bootstrapServers []string,
	groupID string,
	topic string,
	username string,
	password string,
) *kafka.Reader {
	readerConfig := kafka.ReaderConfig{
		// Kafka configure
		Brokers: bootstrapServers,
		// Topic configure
		Topic:                  topic,
		GroupID:                groupID,
		WatchPartitionChanges:  true,
		PartitionWatchInterval: 5 * time.Second,

		// offset configure
		StartOffset: kafka.LastOffset,

		// Batching configure
		MinBytes:         1,
		MaxBytes:         256 * 1024 * 1024, // MiB
		MaxWait:          25 * time.Second,
		QueueCapacity:    1000,
		ReadBatchTimeout: 1 * time.Minute,
		CommitInterval:   time.Millisecond * 500,

		// Logger configure
		Logger:      kafka.LoggerFunc(logging.GetLogger("info")),
		ErrorLogger: kafka.LoggerFunc(logging.GetLogger("error")),

		// Safe default
		MaxAttempts:       3,
		HeartbeatInterval: 3 * time.Second,
	}

	if username != "" && password != "" {
		readerConfig.Dialer = &kafka.Dialer{
			Timeout:   10 * time.Second,
			DualStack: true,
			SASLMechanism: plain.Mechanism{
				Username: username,
				Password: password,
			},
		}
	}

	return kafka.NewReader(readerConfig)
}