How to create a single transport for readers and writers

[lzap]

Hi,

I would like to create a single connection pool (a transport) for all the communication with our Kafka cluster. So I create a kafka.Transport and pass it into kafka.Writer. So far so good, however, when I try to do the same for kafka.Reader there is no configuration field.

Can you explain me what am I missing?

Is it possible that consumers are "always connected" thus connection pooling does not make any sense for them?

Is it safe to create a single dialer and share it similarly to transport?

Here is my code:

// ...

var (
	dialer    *kafka.Dialer
	transport *kafka.Transport
)

func createSaslMechanism(saslMechanismName string, username string, password string) (sasl.Mechanism, error) {
	switch strings.ToLower(saslMechanismName) {
	case "plain":
		return plain.Mechanism{
			Username: username,
			Password: password,
		}, nil
	case "scram-sha-512":
		mechanism, err := scram.Mechanism(scram.SHA512, username, password)
		if err != nil {
			return nil, fmt.Errorf("unable to create scram-sha-512 mechanism: %w", err)
		}

		return mechanism, nil
	case "scram-sha-256":
		mechanism, err := scram.Mechanism(scram.SHA256, username, password)
		if err != nil {
			return nil, fmt.Errorf("unable to create scram-sha-256 mechanism: %w", err)
		}

		return mechanism, nil
	default:
		return nil, fmt.Errorf("unable to configure sasl mechanism (%s)", saslMechanismName)
	}
}

func Initialize() error {
	cleanCAPath := filepath.Clean(config.Kafka.CACert)
	pemCerts, err := ioutil.ReadFile(cleanCAPath)
	if err != nil {
		return fmt.Errorf("unable to read kafka CA cert: %w", err)
	}

	pool := x509.NewCertPool()
	pool.AppendCertsFromPEM(pemCerts)

	tlsConfig := &tls.Config{
		MinVersion: tls.VersionTLS13,
		RootCAs:    pool,
	}

	saslMechanism, err := createSaslMechanism(config.Kafka.SASL.SaslMechanism, config.Kafka.SASL.Username, config.Kafka.SASL.Password)
	if err != nil {
		return fmt.Errorf("kafka SASL error: %w", err)
	}

	dialer = &kafka.Dialer{
		Timeout:       10 * time.Second,
		DualStack:     true,
		SASLMechanism: saslMechanism,
		TLS:           tlsConfig,
	}

	transport = &kafka.Transport{
		Dial:     dialer.DialFunc,
		ClientID: version.KafkaClientID,
		TLS:      tlsConfig,
		SASL:     saslMechanism,
	}
	
	return nil
}

func newContextLogger(ctx context.Context) func(msg string, a ...interface{}) {
	return func(msg string, a ...interface{}) {
		logger := ctxval.Logger(ctx)
		logger.Debug().Bool("kafka", true).Msgf(msg, a...)
	}
}

func newContextErrLogger(ctx context.Context) func(msg string, a ...interface{}) {
	return func(msg string, a ...interface{}) {
		logger := ctxval.Logger(ctx)
		logger.Warn().Bool("kafka", true).Msgf(msg, a...)
	}
}

func NewReader(ctx context.Context, topic string) *kafka.Reader {
	return kafka.NewReader(kafka.ReaderConfig{
		Brokers:     config.Kafka.Brokers,
		Dialer:      dialer,
		Topic:       topic,
		Logger:      kafka.LoggerFunc(newContextLogger(ctx)),
		ErrorLogger: kafka.LoggerFunc(newContextErrLogger(ctx)),
	})
}

func NewWriter(ctx context.Context, topic string) *kafka.Writer {
	return &kafka.Writer{
		Addr:        kafka.TCP(config.Kafka.Brokers...),
		Topic:       topic,
		Transport:   transport,
		Logger:      kafka.LoggerFunc(newContextLogger(ctx)),
		ErrorLogger: kafka.LoggerFunc(newContextErrLogger(ctx)),
	}
}

Thanks!

[Abdulsametileri]

@achille-roussel