client

type AdminConfig ¶

type AdminConfig struct {
	// contains filtered or unexported fields
}

AdminConfig holds the configuration for creating a Kafka admin client

type AdminOption ¶

type AdminOption func(*AdminConfig)

AdminOption is a function type for configuring the Kafka admin client

func WithIAMAuth() AdminOption

func WithSASLSCRAMAuth(username, password string) AdminOption

func WithTLSAuth(caCertFile string, clientCertFile string, clientKeyFile string) AdminOption

func WithUnauthenticatedAuth() AdminOption

type ClusterKafkaMetadata ¶

type ClusterKafkaMetadata struct {
	Brokers      []*sarama.Broker
	ControllerID int32
	ClusterID    string
}

ClusterKafkaMetadata represents cluster information including brokers, controller, and cluster ID

type KafkaAdmin ¶

type KafkaAdmin interface {
	ListTopics() (map[string]sarama.TopicDetail, error)
	GetClusterKafkaMetadata() (*ClusterKafkaMetadata, error)
	DescribeConfig() ([]sarama.ConfigEntry, error)
	ListAcls() ([]sarama.ResourceAcls, error)
	Close() error
}

KafkaAdmin interface defines the Kafka admin operations we need

func NewKafkaAdmin(brokerAddresses []string, clientBrokerEncryptionInTransit kafkatypes.ClientBroker, region string, kafkaVersion string, opts ...AdminOption) (KafkaAdmin, error)

type KafkaAdminClient ¶

type KafkaAdminClient struct {
	// contains filtered or unexported fields
}

KafkaAdminClient wraps sarama.ClusterAdmin to implement our KafkaAdmin interface

func (k *KafkaAdminClient) Close() error

func (k *KafkaAdminClient) DescribeConfig() ([]sarama.ConfigEntry, error)

func (k *KafkaAdminClient) GetClusterKafkaMetadata() (*ClusterKafkaMetadata, error)

func (k *KafkaAdminClient) ListAcls() ([]sarama.ResourceAcls, error)

func (k *KafkaAdminClient) ListTopics() (map[string]sarama.TopicDetail, error)

type MSKAccessTokenProvider ¶

type MSKAccessTokenProvider struct {
	// contains filtered or unexported fields
}

MSKAccessTokenProvider implements sarama.AccessTokenProvider for MSK IAM authentication

func (m *MSKAccessTokenProvider) Token() (*sarama.AccessToken, error)

type XDGSCRAMClient ¶

type XDGSCRAMClient struct {
	*scram.Client
	*scram.ClientConversation
	scram.HashGeneratorFcn
}

func (x *XDGSCRAMClient) Begin(userName, password, authzID string) (err error)

func (x *XDGSCRAMClient) Done() bool

func (x *XDGSCRAMClient) Step(challenge string) (response string, err error)