kafka

package
v1.2.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 3, 2026 License: MIT Imports: 22 Imported by: 0

README

Managed Streaming for Kafka

Parity grade: A · SDK aws-sdk-go-v2/service/kafka@v1.49.0 · last audited 2026-07-23 (fb5f045f5a201fb9817e392cdf36684aa6cb36e6)

Coverage

Metric Value
Operations audited 59 (59 ok)
Feature families 12 (12 ok)
Known gaps none
Deferred items 0
Resource leaks clean

More

Documentation

Overview

Package kafka provides an in-memory stub of AWS MSK (Managed Streaming for Apache Kafka).

Index

Constants

View Source 
const (
	// ReplicatorStateRunning indicates a running replicator.
	ReplicatorStateRunning = "RUNNING"
	// VpcConnectionStateAvailable indicates a VPC connection that is available.
	VpcConnectionStateAvailable = "AVAILABLE"
	// ClusterOperationStateUpdateComplete indicates a completed cluster operation.
	ClusterOperationStateUpdateComplete = "UPDATE_COMPLETE"
	// DefaultClusterVersion is the default MSK cluster version identifier.
	DefaultClusterVersion = "K3AEGXETSR30VB"
)
View Source 
const (
	// ClusterStateActive indicates a running cluster.
	ClusterStateActive = "ACTIVE"
	// ClusterStateCreating indicates a cluster being provisioned.
	ClusterStateCreating = "CREATING"
	// ClusterStateDeleting indicates a cluster being removed.
	ClusterStateDeleting = "DELETING"
	// ClusterStateFailed indicates a cluster in a failed state.
	ClusterStateFailed = "FAILED"
	// ClusterStateUpdating indicates a cluster undergoing an update.
	ClusterStateUpdating = "UPDATING"
	// ClusterStateRebootingBroker indicates a broker reboot in progress.
	ClusterStateRebootingBroker = "REBOOTING_BROKER"
	// ClusterStateMaintenance indicates a cluster in a maintenance window.
	ClusterStateMaintenance = "MAINTENANCE"
	// ClusterStateHealing indicates a cluster undergoing node healing.
	ClusterStateHealing = "HEALING"
)
View Source 
const (
	// ClusterTypeProvisioned is the standard MSK cluster type.
	ClusterTypeProvisioned = "PROVISIONED"
	// ClusterTypeServerless is the MSK Serverless cluster type.
	ClusterTypeServerless = "SERVERLESS"
)
View Source 
const (
	// EnhancedMonitoringDefault is the default monitoring level.
	EnhancedMonitoringDefault = "DEFAULT"
	// EnhancedMonitoringPerBroker enables per-broker metrics.
	EnhancedMonitoringPerBroker = "PER_BROKER"
	// EnhancedMonitoringPerTopicPerBroker enables per-topic-per-broker metrics.
	EnhancedMonitoringPerTopicPerBroker = "PER_TOPIC_PER_BROKER"
	// EnhancedMonitoringPerTopicPerPartition enables per-topic-per-partition metrics.
	EnhancedMonitoringPerTopicPerPartition = "PER_TOPIC_PER_PARTITION"
)
View Source 
const (
	// StorageModeLocal is the standard EBS-only storage mode.
	StorageModeLocal = "LOCAL"
	// StorageModeTiered enables tiered storage (remote storage offload).
	StorageModeTiered = "TIERED"
)
View Source 
const (
	// TopicStateActive indicates a topic ready for use.
	TopicStateActive = "ACTIVE"
	// TopicStateCreating indicates a topic being created.
	TopicStateCreating = "CREATING"
	// TopicStateUpdating indicates a topic undergoing a configuration update.
	TopicStateUpdating = "UPDATING"
	// TopicStateDeleting indicates a topic being removed.
	TopicStateDeleting = "DELETING"
)
View Source 
const (
	// EncryptionInTransitTLS requires TLS for all client-broker traffic.
	EncryptionInTransitTLS = "TLS"
	// EncryptionInTransitTLSPlaintext allows both TLS and plaintext.
	EncryptionInTransitTLSPlaintext = "TLS_PLAINTEXT"
	// EncryptionInTransitPlaintext allows plaintext only (dev/test use).
	EncryptionInTransitPlaintext = "PLAINTEXT"
)

Variables

View Source 
var (
	// ErrNotFound is returned when a requested resource does not exist.
	ErrNotFound = awserr.New("NotFoundException", awserr.ErrNotFound)
	// ErrAlreadyExists is returned when a resource already exists.
	ErrAlreadyExists = awserr.New("ConflictException", awserr.ErrAlreadyExists)
	// ErrValidation is returned when input validation fails.
	ErrValidation = awserr.New("BadRequestException", awserr.ErrInvalidParameter)
)
View Source 
var ErrNilAppContext = errors.New("nil AppContext passed to Kafka Provider.Init")

ErrNilAppContext is returned by Init when a nil AppContext is passed.

Functions

This section is empty.

Types

type BrokerEBSVolumeInfo 

type BrokerEBSVolumeInfo struct {
	ProvisionedThroughput *ProvisionedThroughput `json:"provisionedThroughput,omitempty"`
	KafkaBrokerNodeID     string                 `json:"kafkaBrokerNodeId,omitempty"`
	VolumeSizeGB          int32                  `json:"volumeSizeGB,omitempty"`
}

BrokerEBSVolumeInfo describes a per-broker EBS volume target for UpdateStorage.

type BrokerLogs 

type BrokerLogs struct {
	CloudWatchLogs *CloudWatchLogs `json:"cloudWatchLogs,omitempty"`
	Firehose       *Firehose       `json:"firehose,omitempty"`
	S3             *S3Logs         `json:"s3,omitempty"`
}

BrokerLogs holds broker log delivery destinations.

type BrokerNode 

type BrokerNode struct {
	InstanceType string `json:"instanceType,omitempty"`
	BrokerID     int32  `json:"brokerId"`
}

BrokerNode represents a stub broker node.

type BrokerNodeGroupInfo 

type BrokerNodeGroupInfo struct {
	ConnectivityInfo     *ConnectivityInfo `json:"connectivityInfo,omitempty"`
	StorageInfo          *StorageInfo      `json:"storageInfo,omitempty"`
	BrokerAZDistribution string            `json:"brokerAZDistribution,omitempty"`
	InstanceType         string            `json:"instanceType"`
	ZoneIDs              []string          `json:"zoneIds,omitempty"`
	ClientSubnets        []string          `json:"clientSubnets"`
	SecurityGroups       []string          `json:"securityGroups,omitempty"`
}

BrokerNodeGroupInfo holds broker node configuration.

type ClientAuthentication 

type ClientAuthentication struct {
	Sasl            *SaslSettings            `json:"sasl,omitempty"`
	TLS             *TLSSettings             `json:"tls,omitempty"`
	Unauthenticated *UnauthenticatedSettings `json:"unauthenticated,omitempty"`
}

ClientAuthentication holds MSK cluster authentication configuration.

type CloudWatchLogs 

type CloudWatchLogs struct {
	LogGroup string `json:"logGroup,omitempty"`
	Enabled  bool   `json:"enabled"`
}

CloudWatchLogs holds CloudWatch log delivery settings.

type Cluster 

type Cluster struct {
	Tags                 map[string]string      `json:"-"`
	ClientAuthentication *ClientAuthentication  `json:"clientAuthentication,omitempty"`
	EncryptionInfo       *EncryptionInfo        `json:"encryptionInfo,omitempty"`
	OpenMonitoring       *OpenMonitoring        `json:"openMonitoring,omitempty"`
	LoggingInfo          *LoggingInfo           `json:"loggingInfo,omitempty"`
	StateInfo            *StateInfo             `json:"stateInfo,omitempty"`
	Serverless           *ServerlessClusterInfo `json:"serverless,omitempty"`
	ConfigurationInfo    *ConfigurationInfo     `json:"configurationInfo,omitempty"`
	ClusterArn           string                 `json:"clusterArn"`
	ClusterName          string                 `json:"clusterName"`
	ClusterType          string                 `json:"clusterType"`
	KafkaVersion         string                 `json:"kafkaVersion,omitempty"`
	State                string                 `json:"state"`
	CurrentVersion       string                 `json:"currentVersion"`
	ActiveOperationArn   string                 `json:"activeOperationArn,omitempty"`
	EnhancedMonitoring   string                 `json:"enhancedMonitoring,omitempty"`
	StorageMode          string                 `json:"storageMode,omitempty"`
	CreationTime         string                 `json:"creationTime,omitempty"`
	BrokerNodeGroupInfo  BrokerNodeGroupInfo    `json:"brokerNodeGroupInfo"`
	NumberOfBrokerNodes  int32                  `json:"numberOfBrokerNodes"`
	// contains filtered or unexported fields
}

Cluster represents an MSK cluster.

type ClusterConfig added in v1.2.0 

type ClusterConfig struct {
	MskClusterArn    string   `json:"mskClusterArn"`
	Alias            string   `json:"-"`
	SubnetIDs        []string `json:"subnetIds,omitempty"`
	SecurityGroupIDs []string `json:"securityGroupIds,omitempty"`
}

ClusterConfig describes one MSK cluster referenced by a replicator's kafkaClusters list (the Create/Describe "source or target" side), mirroring aws-sdk-go-v2/service/kafka/types.KafkaCluster's amazonMskCluster+vpcConfig shape. Alias is never set on the persisted record -- it is resolved fresh on every read (see InMemoryBackend.clusterAliasForArn) to mirror types. KafkaClusterDescription.KafkaClusterAlias, which real MSK derives from the referenced cluster's current name rather than storing it at creation time.

type ClusterOperation 

type ClusterOperation struct {
	SourceClusterInfo   *MutableClusterInfo `json:"sourceClusterInfo,omitempty"`
	TargetClusterInfo   *MutableClusterInfo `json:"targetClusterInfo,omitempty"`
	ClusterOperationArn string              `json:"clusterOperationArn"`
	ClusterArn          string              `json:"clusterArn"`
	OperationType       string              `json:"operationType"`
	OperationState      string              `json:"operationState"`
}

ClusterOperation represents an MSK cluster operation.

type Configuration 

type Configuration struct {
	Tags             map[string]string `json:"-"`
	Arn              string            `json:"arn"`
	Name             string            `json:"name"`
	Description      string            `json:"description,omitempty"`
	ServerProperties string            `json:"serverProperties"`
	KafkaVersions    []string          `json:"kafkaVersions"`
}

Configuration represents an MSK configuration.

type ConfigurationInfo 

type ConfigurationInfo struct {
	Arn      string `json:"arn"`
	Revision int64  `json:"revision"`
}

ConfigurationInfo holds a cluster configuration reference.

type ConfigurationRevision 

type ConfigurationRevision struct {
	ConfigurationArn string `json:"configurationArn"`
	Description      string `json:"description,omitempty"`
	ServerProperties string `json:"serverProperties,omitempty"`
	Revision         int64  `json:"revision"`
}

ConfigurationRevision represents a revision of an MSK configuration.

type ConnectivityInfo 

type ConnectivityInfo struct {
	PublicAccess    *PublicAccess    `json:"publicAccess,omitempty"`
	VpcConnectivity *VpcConnectivity `json:"vpcConnectivity,omitempty"`
}

ConnectivityInfo holds broker connectivity configuration.

type ConsumerGroupReplicationConfig added in v1.2.0 

type ConsumerGroupReplicationConfig struct {
	ConsumerGroupsToExclude         []string `json:"consumerGroupsToExclude,omitempty"`
	ConsumerGroupsToReplicate       []string `json:"consumerGroupsToReplicate,omitempty"`
	DetectAndCopyNewConsumerGroups  bool     `json:"detectAndCopyNewConsumerGroups"`
	SynchroniseConsumerGroupOffsets bool     `json:"synchroniseConsumerGroupOffsets"`
}

ConsumerGroupReplicationConfig mirrors types.ConsumerGroupReplication / types.ConsumerGroupReplicationUpdate.

type EBSStorageInfo 

type EBSStorageInfo struct {
	ProvisionedThroughput *ProvisionedThroughput `json:"provisionedThroughput,omitempty"`
	VolumeSize            int32                  `json:"volumeSize,omitempty"`
}

EBSStorageInfo holds EBS volume config.

type EncryptionAtRest 

type EncryptionAtRest struct {
	DataVolumeKMSKeyID string `json:"dataVolumeKMSKeyId,omitempty"`
}

EncryptionAtRest holds at-rest encryption configuration.

type EncryptionInTransit 

type EncryptionInTransit struct {
	ClientBroker string `json:"clientBroker,omitempty"`
	InCluster    bool   `json:"inCluster"`
}

EncryptionInTransit holds in-transit encryption configuration.

type EncryptionInfo 

type EncryptionInfo struct {
	EncryptionAtRest    *EncryptionAtRest    `json:"encryptionAtRest,omitempty"`
	EncryptionInTransit *EncryptionInTransit `json:"encryptionInTransit,omitempty"`
}

EncryptionInfo holds cluster encryption configuration.

type Firehose 

type Firehose struct {
	DeliveryStream string `json:"deliveryStream,omitempty"`
	Enabled        bool   `json:"enabled"`
}

Firehose holds Kinesis Data Firehose log delivery settings.

type Handler 

type Handler struct {
	Backend StorageBackend
}

Handler is the HTTP handler for the MSK REST API.

func NewHandler 

func NewHandler(backend StorageBackend) *Handler

NewHandler creates a new Kafka handler backed by backend.

func (*Handler) ChaosOperations 

func (h *Handler) ChaosOperations() []string

ChaosOperations returns all operations that can be fault-injected.

func (*Handler) ChaosRegions 

func (h *Handler) ChaosRegions() []string

ChaosRegions returns all regions this handler instance handles.

func (*Handler) ChaosServiceName 

func (h *Handler) ChaosServiceName() string

ChaosServiceName returns the lowercase AWS service name for fault rule matching.

func (*Handler) ExtractOperation 

func (h *Handler) ExtractOperation(c *echo.Context) string

ExtractOperation extracts the MSK operation name from the request.

func (*Handler) ExtractResource 

func (h *Handler) ExtractResource(c *echo.Context) string

ExtractResource extracts a resource ARN from the URL path.

func (*Handler) GetSupportedOperations 

func (h *Handler) GetSupportedOperations() []string

GetSupportedOperations returns the list of supported MSK operations.

func (*Handler) Handler 

func (h *Handler) Handler() echo.HandlerFunc

Handler returns the Echo handler function for MSK requests.

func (*Handler) MatchPriority 

func (h *Handler) MatchPriority() int

MatchPriority returns the routing priority.

func (*Handler) Name 

func (h *Handler) Name() string

Name returns the service name.

func (*Handler) Reset 

func (h *Handler) Reset()

Reset clears all backend state.

func (*Handler) Restore 

func (h *Handler) Restore(ctx context.Context, data []byte) error

Restore implements persistence.Persistable by delegating to the backend.

func (*Handler) RouteMatcher 

func (h *Handler) RouteMatcher() service.Matcher

RouteMatcher returns a function that matches MSK REST API requests.

func (*Handler) Snapshot 

func (h *Handler) Snapshot(ctx context.Context) []byte

Snapshot implements persistence.Persistable by delegating to the backend.

type InMemoryBackend 

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

InMemoryBackend stores MSK state in memory.

Every ARN-keyed resource table below is keyed by the resource's own ARN, which already encodes its region (arn:partition:service:region:account: resource, see regionFromARN) — so same-named resources in different regions are fully isolated without an outer region-keyed map layer. Operations that take an existing resource ARN resolve their region from the ARN itself when they need to build a NEW related ARN; point lookups by an existing ARN need no region resolution at all. See store_setup.go for the full table/index registration and the rationale for each one.

func NewInMemoryBackend 

func NewInMemoryBackend(accountID, region string) *InMemoryBackend

NewInMemoryBackend creates a new in-memory MSK backend.

func (*InMemoryBackend) AccountID 

func (b *InMemoryBackend) AccountID() string

AccountID returns the backend account ID.

func (*InMemoryBackend) AddClusterInternal 

func (b *InMemoryBackend) AddClusterInternal(name, kafkaVersion string) *Cluster

AddClusterInternal creates a cluster directly for testing purposes.

func (*InMemoryBackend) AddClusterOperationInternal 

func (b *InMemoryBackend) AddClusterOperationInternal(
	clusterArn, operationType string,
) *ClusterOperation

AddClusterOperationInternal adds a cluster operation for testing purposes.

func (*InMemoryBackend) AddConfigurationInternal 

func (b *InMemoryBackend) AddConfigurationInternal(name string) *Configuration

func (*InMemoryBackend) AddReplicatorInternal 

func (b *InMemoryBackend) AddReplicatorInternal(name string) *Replicator

AddReplicatorInternal creates a replicator directly for testing purposes.

func (*InMemoryBackend) AddTopicInternal 

func (b *InMemoryBackend) AddTopicInternal(clusterArn, topicName string) *Topic

AddTopicInternal creates a topic directly for testing purposes.

func (*InMemoryBackend) AddVpcConnectionInternal 

func (b *InMemoryBackend) AddVpcConnectionInternal(clusterArn, vpcID string) *VpcConnection

AddVpcConnectionInternal creates a VPC connection directly for testing purposes.

func (*InMemoryBackend) BatchAssociateScramSecret 

func (b *InMemoryBackend) BatchAssociateScramSecret(
	_ context.Context,
	clusterArn string,
	secretArnList []string,
) ([]ScramSecretError, error)

BatchAssociateScramSecret associates a list of SCRAM secrets with a cluster. It returns any errors that occurred for individual secrets.

func (*InMemoryBackend) BatchDisassociateScramSecret 

func (b *InMemoryBackend) BatchDisassociateScramSecret(
	_ context.Context,
	clusterArn string,
	secretArnList []string,
) ([]ScramSecretError, error)

BatchDisassociateScramSecret disassociates a list of SCRAM secrets from a cluster. It returns any errors that occurred for individual secrets.

func (*InMemoryBackend) CreateCluster 

func (b *InMemoryBackend) CreateCluster(
	ctx context.Context,
	name, kafkaVersion string,
	numBrokers int32,
	brokerInfo BrokerNodeGroupInfo,
	clientAuth *ClientAuthentication,
	tags map[string]string,
) (*Cluster, error)

CreateCluster creates a new MSK cluster.

func (*InMemoryBackend) CreateConfiguration 

func (b *InMemoryBackend) CreateConfiguration(
	ctx context.Context,
	name, description string,
	kafkaVersions []string,
	serverProperties string,
) (*Configuration, error)

CreateConfiguration creates a new MSK configuration.

func (*InMemoryBackend) CreateReplicator 

func (b *InMemoryBackend) CreateReplicator(
	ctx context.Context,
	name, description, serviceExecutionRoleArn string,
	kafkaClusters []ClusterConfig,
	replicationInfoList []ReplicationInfoConfig,
	tags map[string]string,
) (*Replicator, error)

CreateReplicator creates a new MSK replicator. Real aws-sdk-go-v2 clients always populate kafkaClusters/replicationInfoList (the SDK's client-side input-validation middleware rejects a CreateReplicatorInput missing either before the request is even sent), so both are accepted and fully persisted here but not hard-required server-side -- a nil/empty slice from a non-SDK caller is stored as-is rather than rejected.

func (*InMemoryBackend) CreateServerlessCluster 

func (b *InMemoryBackend) CreateServerlessCluster(
	ctx context.Context,
	name string,
	serverless *ServerlessClusterInfo,
	tags map[string]string,
) (*Cluster, error)

CreateServerlessCluster creates a new MSK Serverless cluster.

func (*InMemoryBackend) CreateTopic 

func (b *InMemoryBackend) CreateTopic(
	_ context.Context,
	clusterArn, topicName string,
	replicationFactor, partitionCount int32,
	configs string,
) (*Topic, error)

CreateTopic creates a topic on an MSK cluster. Real MSK creates topics synchronously from the caller's perspective (no polling protocol is exposed for topic creation the way Cluster CREATING->ACTIVE is), so the topic is ACTIVE immediately.

func (*InMemoryBackend) CreateVpcConnection 

func (b *InMemoryBackend) CreateVpcConnection(
	ctx context.Context,
	targetClusterArn, vpcID, authentication string,
	clientSubnets, securityGroups []string,
	tags map[string]string,
) (*VpcConnection, error)

CreateVpcConnection creates a new VPC connection to an MSK cluster. clientSubnets/securityGroups are required fields on the real CreateVpcConnectionInput (see aws-sdk-go-v2/service/kafka's api_op_CreateVpcConnection.go); they are stored and echoed back by DescribeVpcConnection.

func (*InMemoryBackend) DeleteCluster 

func (b *InMemoryBackend) DeleteCluster(_ context.Context, clusterArn string) error

DeleteCluster deletes a cluster by ARN, cascading to its SCRAM secrets, topics and cluster policy.

func (*InMemoryBackend) DeleteClusterPolicy 

func (b *InMemoryBackend) DeleteClusterPolicy(_ context.Context, clusterArn string) error

DeleteClusterPolicy deletes the policy attached to an MSK cluster.

func (*InMemoryBackend) DeleteConfiguration 

func (b *InMemoryBackend) DeleteConfiguration(_ context.Context, configArn string) error

DeleteConfiguration deletes a configuration by ARN.

func (*InMemoryBackend) DeleteReplicator 

func (b *InMemoryBackend) DeleteReplicator(_ context.Context, replicatorArn string) error

DeleteReplicator deletes a replicator by ARN.

func (*InMemoryBackend) DeleteTopic 

func (b *InMemoryBackend) DeleteTopic(_ context.Context, clusterArn, topicName string) error

DeleteTopic deletes a topic from an MSK cluster.

func (*InMemoryBackend) DeleteVpcConnection 

func (b *InMemoryBackend) DeleteVpcConnection(_ context.Context, vpcConnectionArn string) error

DeleteVpcConnection deletes a VPC connection by ARN.

func (*InMemoryBackend) DescribeCluster 

func (b *InMemoryBackend) DescribeCluster(_ context.Context, clusterArn string) (*Cluster, error)

DescribeCluster retrieves a cluster by ARN, advancing CREATING→ACTIVE on first poll.

func (*InMemoryBackend) DescribeClusterOperation 

func (b *InMemoryBackend) DescribeClusterOperation(
	_ context.Context,
	clusterOperationArn string,
) (*ClusterOperation, error)

DescribeClusterOperation retrieves a cluster operation by ARN.

func (*InMemoryBackend) DescribeClusterOperationV2 

func (b *InMemoryBackend) DescribeClusterOperationV2(
	ctx context.Context,
	clusterOperationArn string,
) (*ClusterOperation, error)

DescribeClusterOperationV2 retrieves a cluster operation (V2) by ARN.

func (*InMemoryBackend) DescribeConfiguration 

func (b *InMemoryBackend) DescribeConfiguration(_ context.Context, configArn string) (*Configuration, error)

DescribeConfiguration retrieves a configuration by ARN.

func (*InMemoryBackend) DescribeConfigurationRevision 

func (b *InMemoryBackend) DescribeConfigurationRevision(
	_ context.Context,
	configArn string,
	revision int64,
) (*ConfigurationRevision, error)

DescribeConfigurationRevision retrieves a configuration revision. In this stub, revision 1 always refers to the current configuration state.

func (*InMemoryBackend) DescribeReplicator 

func (b *InMemoryBackend) DescribeReplicator(_ context.Context, replicatorArn string) (*Replicator, error)

DescribeReplicator retrieves a replicator by ARN.

func (*InMemoryBackend) DescribeTopic 

func (b *InMemoryBackend) DescribeTopic(_ context.Context, clusterArn, topicName string) (*Topic, error)

DescribeTopic retrieves a topic by cluster ARN and topic name.

func (*InMemoryBackend) DescribeTopicPartitions 

func (b *InMemoryBackend) DescribeTopicPartitions(
	_ context.Context, clusterArn, topicName string,
) ([]*TopicPartitionInfo, error)

DescribeTopicPartitions synthesizes per-partition placement info for a topic: for partition i, the leader/replica set is a round-robin assignment over the owning cluster's broker IDs (1..NumberOfBrokerNodes), with the full replica set reported in-sync (this in-memory emulator has no real broker/ISR state to diverge from). Returns ErrNotFound if the cluster or topic doesn't exist.

func (*InMemoryBackend) DescribeVpcConnection 

func (b *InMemoryBackend) DescribeVpcConnection(_ context.Context, vpcConnectionArn string) (*VpcConnection, error)

DescribeVpcConnection retrieves a VPC connection by ARN.

func (*InMemoryBackend) GetClusterPolicy 

func (b *InMemoryBackend) GetClusterPolicy(_ context.Context, clusterArn string) (string, error)

GetClusterPolicy retrieves the policy document for a cluster. Returns ErrNotFound when the cluster exists but has no policy set — matching AWS behavior.

func (*InMemoryBackend) GetCompatibleKafkaVersions 

func (b *InMemoryBackend) GetCompatibleKafkaVersions(_ context.Context, clusterArn string) ([]*MSKVersion, error)

GetCompatibleKafkaVersions returns Kafka versions compatible with the cluster's current version. KRaft clusters can only target KRaft versions. ZooKeeper clusters can target ZooKeeper versions up to 3.x.

func (*InMemoryBackend) GetTags 

func (b *InMemoryBackend) GetTags(_ context.Context, resourceArn string) (map[string]string, error)

GetTags retrieves tags for a cluster, configuration, replicator, or VPC connection by ARN.

func (*InMemoryBackend) ListClientVpcConnections 

func (b *InMemoryBackend) ListClientVpcConnections(_ context.Context, clusterArn string) ([]*VpcConnection, error)

ListClientVpcConnections returns all VPC connections for a given cluster.

func (*InMemoryBackend) ListClusterOperations 

func (b *InMemoryBackend) ListClusterOperations(_ context.Context, clusterArn string) ([]*ClusterOperation, error)

ListClusterOperations returns all cluster operations for a cluster.

func (*InMemoryBackend) ListClusterOperationsV2 

func (b *InMemoryBackend) ListClusterOperationsV2(ctx context.Context, clusterArn string) ([]*ClusterOperation, error)

ListClusterOperationsV2 returns all cluster operations for a cluster (V2).

func (*InMemoryBackend) ListClusters 

func (b *InMemoryBackend) ListClusters(ctx context.Context) []*Cluster

ListClusters returns all MSK clusters in the request's region sorted by name.

func (*InMemoryBackend) ListConfigurationRevisions 

func (b *InMemoryBackend) ListConfigurationRevisions(
	_ context.Context,
	configArn string,
) ([]*ConfigurationRevision, error)

ListConfigurationRevisions lists revisions for a configuration. In this stub, every configuration has a single revision (revision 1).

func (*InMemoryBackend) ListConfigurations 

func (b *InMemoryBackend) ListConfigurations(ctx context.Context) []*Configuration

ListConfigurations returns all MSK configurations in the request's region sorted by name.

func (*InMemoryBackend) ListKafkaVersions 

func (b *InMemoryBackend) ListKafkaVersions(_ context.Context) []*MSKVersion

ListKafkaVersions returns supported Kafka versions, matching current MSK availability. Kafka versions are global (not region-scoped), so ctx is unused.

func (*InMemoryBackend) ListNodes 

func (b *InMemoryBackend) ListNodes(_ context.Context, clusterArn string) ([]*BrokerNode, error)

ListNodes returns broker node stubs for a cluster.

func (*InMemoryBackend) ListReplicators 

func (b *InMemoryBackend) ListReplicators(ctx context.Context) []*Replicator

ListReplicators returns all replicators in the request's region sorted by name.

func (*InMemoryBackend) ListScramSecrets 

func (b *InMemoryBackend) ListScramSecrets(_ context.Context, clusterArn string) ([]string, error)

ListScramSecrets returns all SCRAM secrets for a cluster.

func (*InMemoryBackend) ListTopics 

func (b *InMemoryBackend) ListTopics(_ context.Context, clusterArn, nameFilter string) ([]*Topic, error)

ListTopics returns topics for a cluster sorted by topic name, optionally filtered to those whose name starts with nameFilter (matching real MSK's topicNameFilter query parameter; an empty nameFilter matches everything).

func (*InMemoryBackend) ListVpcConnections 

func (b *InMemoryBackend) ListVpcConnections(ctx context.Context) []*VpcConnection

ListVpcConnections returns all VPC connections in the request's region sorted by ARN.

func (*InMemoryBackend) PutClusterPolicy 

func (b *InMemoryBackend) PutClusterPolicy(_ context.Context, clusterArn, policy string) error

PutClusterPolicy sets the policy document for a cluster.

func (*InMemoryBackend) RebootBroker 

func (b *InMemoryBackend) RebootBroker(ctx context.Context, clusterArn string, _ []string) (*ClusterOperation, error)

RebootBroker initiates a broker reboot operation.

func (*InMemoryBackend) Region 

func (b *InMemoryBackend) Region() string

Region returns the backend region.

func (*InMemoryBackend) RejectClientVpcConnection 

func (b *InMemoryBackend) RejectClientVpcConnection(ctx context.Context, vpcConnectionArn string) error

RejectClientVpcConnection rejects (deletes) a VPC connection.

func (*InMemoryBackend) Reset 

func (b *InMemoryBackend) Reset()

Reset clears all state, returning the backend to a clean empty state.

func (*InMemoryBackend) Restore 

func (b *InMemoryBackend) Restore(ctx context.Context, data []byte) error

Restore loads backend state from a JSON snapshot. It implements persistence.Persistable.

func (*InMemoryBackend) Snapshot 

func (b *InMemoryBackend) Snapshot(ctx context.Context) []byte

Snapshot serialises the backend state to JSON. It implements persistence.Persistable.

func (*InMemoryBackend) TagResource 

func (b *InMemoryBackend) TagResource(_ context.Context, resourceArn string, tags map[string]string) error

TagResource adds tags to a cluster, configuration, replicator, or VPC connection by ARN.

func (*InMemoryBackend) UntagResource 

func (b *InMemoryBackend) UntagResource(_ context.Context, resourceArn string, tagKeys []string) error

UntagResource removes tags from a cluster, configuration, replicator, or VPC connection by ARN.

func (*InMemoryBackend) UpdateBrokerCount 

func (b *InMemoryBackend) UpdateBrokerCount(
	ctx context.Context,
	clusterArn string,
	numBrokers int32,
) (*ClusterOperation, error)

UpdateBrokerCount updates the number of broker nodes in a cluster.

func (*InMemoryBackend) UpdateBrokerStorage 

func (b *InMemoryBackend) UpdateBrokerStorage(
	ctx context.Context,
	clusterArn string,
	volumeSize int32,
) (*ClusterOperation, error)

UpdateBrokerStorage updates the EBS storage size for broker nodes.

func (*InMemoryBackend) UpdateBrokerType 

func (b *InMemoryBackend) UpdateBrokerType(
	ctx context.Context,
	clusterArn, instanceType string,
) (*ClusterOperation, error)

UpdateBrokerType updates the instance type for broker nodes.

func (*InMemoryBackend) UpdateClusterConfiguration 

func (b *InMemoryBackend) UpdateClusterConfiguration(
	ctx context.Context,
	clusterArn, configArn string,
	revision int64,
) (*ClusterOperation, error)

UpdateClusterConfiguration updates the configuration for a cluster.

func (*InMemoryBackend) UpdateClusterKafkaVersion 

func (b *InMemoryBackend) UpdateClusterKafkaVersion(
	ctx context.Context,
	clusterArn, targetKafkaVersion string,
) (*ClusterOperation, error)

UpdateClusterKafkaVersion updates the Kafka version for a cluster.

func (*InMemoryBackend) UpdateConfiguration 

func (b *InMemoryBackend) UpdateConfiguration(
	_ context.Context,
	configArn, description, serverProperties string,
) (*Configuration, error)

UpdateConfiguration updates a configuration's server properties and description.

func (*InMemoryBackend) UpdateConnectivity 

func (b *InMemoryBackend) UpdateConnectivity(
	ctx context.Context,
	clusterArn string, settings UpdateConnectivitySettings,
) (*ClusterOperation, error)

UpdateConnectivity updates broker connectivity settings for a cluster, persisting the new ConnectivityInfo onto the broker node group and recording an operation whose source/target reflect the before/after state.

func (*InMemoryBackend) UpdateMonitoring 

func (b *InMemoryBackend) UpdateMonitoring(
	ctx context.Context,
	clusterArn string, settings UpdateMonitoringSettings,
) (*ClusterOperation, error)

UpdateMonitoring updates monitoring/logging settings for a cluster, persisting the new EnhancedMonitoring/OpenMonitoring/LoggingInfo and recording an operation.

func (*InMemoryBackend) UpdateRebalancing 

func (b *InMemoryBackend) UpdateRebalancing(ctx context.Context, clusterArn string) (*ClusterOperation, error)

UpdateRebalancing records a rebalancing operation for a cluster. AWS MSK exposes no per-field rebalancing configuration to persist (it is an action, not a setting), so this validates the cluster and records the operation.

func (*InMemoryBackend) UpdateReplicationInfo 

func (b *InMemoryBackend) UpdateReplicationInfo(
	_ context.Context,
	replicatorArn, currentVersion, sourceKafkaClusterArn, targetKafkaClusterArn string,
	topicReplication *TopicReplicationConfig,
	consumerGroupReplication *ConsumerGroupReplicationConfig,
) (*Replicator, error)

UpdateReplicationInfo updates the topic/consumer-group replication settings for one source->target flow of a replicator, enforcing the same optimistic-lock CurrentVersion contract as the cluster Update* operations.

func (*InMemoryBackend) UpdateSecurity 

func (b *InMemoryBackend) UpdateSecurity(
	ctx context.Context,
	clusterArn string, settings UpdateSecuritySettings,
) (*ClusterOperation, error)

UpdateSecurity updates authentication/encryption settings for a cluster, persisting the new ClientAuthentication/EncryptionInfo and recording an operation.

func (*InMemoryBackend) UpdateStorage 

func (b *InMemoryBackend) UpdateStorage(
	ctx context.Context,
	clusterArn string, settings UpdateStorageSettings,
) (*ClusterOperation, error)

UpdateStorage updates broker storage settings for a cluster, persisting the new StorageMode and EBS volume size/throughput and recording an operation.

func (*InMemoryBackend) UpdateTopic 

func (b *InMemoryBackend) UpdateTopic(
	_ context.Context,
	clusterArn, topicName string,
	partitionCount int32,
	configs string,
) (*Topic, error)

UpdateTopic updates a topic's configs and/or partition count.

type JmxExporter 

type JmxExporter struct {
	EnabledInBroker bool `json:"enabledInBroker"`
}

JmxExporter holds JMX exporter settings.

type LoggingInfo 

type LoggingInfo struct {
	BrokerLogs *BrokerLogs `json:"brokerLogs,omitempty"`
}

LoggingInfo holds cluster logging configuration.

type MSKVersion 

type MSKVersion struct {
	Version string `json:"version"`
	Status  string `json:"status"`
}

MSKVersion represents an available Kafka version.

type MutableClusterInfo 

type MutableClusterInfo struct {
	ConnectivityInfo     *ConnectivityInfo     `json:"connectivityInfo,omitempty"`
	OpenMonitoring       *OpenMonitoring       `json:"openMonitoring,omitempty"`
	LoggingInfo          *LoggingInfo          `json:"loggingInfo,omitempty"`
	ClientAuthentication *ClientAuthentication `json:"clientAuthentication,omitempty"`
	EncryptionInfo       *EncryptionInfo       `json:"encryptionInfo,omitempty"`
	StorageMode          string                `json:"storageMode,omitempty"`
	EnhancedMonitoring   string                `json:"enhancedMonitoring,omitempty"`
	BrokerEBSVolumeInfo  []BrokerEBSVolumeInfo `json:"brokerEBSVolumeInfo,omitempty"`
	NumberOfBrokerNodes  int32                 `json:"numberOfBrokerNodes,omitempty"`
}

MutableClusterInfo captures the subset of cluster configuration that an update operation changes. DescribeClusterOperation returns it as SourceClusterInfo (the state before the operation) and TargetClusterInfo (the requested state).

type NodeExporter 

type NodeExporter struct {
	EnabledInBroker bool `json:"enabledInBroker"`
}

NodeExporter holds Node exporter settings.

type OpenMonitoring 

type OpenMonitoring struct {
	Prometheus *PrometheusInfo `json:"prometheus,omitempty"`
}

OpenMonitoring holds open monitoring configuration.

type PrometheusInfo 

type PrometheusInfo struct {
	JmxExporter  *JmxExporter  `json:"jmxExporter,omitempty"`
	NodeExporter *NodeExporter `json:"nodeExporter,omitempty"`
}

PrometheusInfo holds Prometheus scraping configuration.

type Provider 

type Provider struct{}

Provider implements service.Provider for MSK (Kafka).

func (*Provider) Init 

func (p *Provider) Init(ctx *service.AppContext) (service.Registerable, error)

Init initializes the Kafka backend and handler.

func (*Provider) Name 

func (p *Provider) Name() string

Name returns the provider name.

type ProvisionedThroughput 

type ProvisionedThroughput struct {
	VolumeThroughput int32 `json:"volumeThroughput,omitempty"`
	Enabled          bool  `json:"enabled"`
}

ProvisionedThroughput holds EBS provisioned throughput config.

type PublicAccess 

type PublicAccess struct {
	Type string `json:"type,omitempty"`
}

PublicAccess holds public access configuration for broker connectivity.

type ReplicationInfoConfig added in v1.2.0 

type ReplicationInfoConfig struct {
	SourceKafkaClusterArn    string                         `json:"sourceKafkaClusterArn"`
	TargetKafkaClusterArn    string                         `json:"targetKafkaClusterArn"`
	TargetCompressionType    string                         `json:"targetCompressionType,omitempty"`
	SourceAlias              string                         `json:"-"`
	TargetAlias              string                         `json:"-"`
	TopicReplication         TopicReplicationConfig         `json:"topicReplication"`
	ConsumerGroupReplication ConsumerGroupReplicationConfig `json:"consumerGroupReplication"`
}

ReplicationInfoConfig mirrors types.ReplicationInfo: one source-cluster to target-cluster replication flow within a replicator. SourceAlias/ TargetAlias are resolved fresh on every read, like ClusterConfig.Alias.

type Replicator 

type Replicator struct {
	Tags                    map[string]string       `json:"-"`
	ReplicatorArn           string                  `json:"replicatorArn"`
	ReplicatorName          string                  `json:"replicatorName"`
	Description             string                  `json:"description,omitempty"`
	ServiceExecutionRoleArn string                  `json:"serviceExecutionRoleArn"`
	ReplicatorState         string                  `json:"replicatorState"`
	CurrentVersion          string                  `json:"currentVersion,omitempty"`
	CreationTime            string                  `json:"creationTime,omitempty"`
	StateInfoCode           string                  `json:"stateInfoCode,omitempty"`
	StateInfoMessage        string                  `json:"stateInfoMessage,omitempty"`
	KafkaClusters           []ClusterConfig         `json:"kafkaClusters,omitempty"`
	ReplicationInfoList     []ReplicationInfoConfig `json:"replicationInfoList,omitempty"`
}

Replicator represents an MSK replicator.

type S3Logs 

type S3Logs struct {
	Bucket  string `json:"bucket,omitempty"`
	Prefix  string `json:"prefix,omitempty"`
	Enabled bool   `json:"enabled"`
}

S3Logs holds S3 log delivery settings.

type SaslIam 

type SaslIam struct {
	Enabled bool `json:"enabled"`
}

SaslIam holds SASL/IAM settings.

type SaslScram 

type SaslScram struct {
	Enabled bool `json:"enabled"`
}

SaslScram holds SASL/SCRAM settings.

type SaslSettings 

type SaslSettings struct {
	Scram *SaslScram `json:"scram,omitempty"`
	Iam   *SaslIam   `json:"iam,omitempty"`
}

SaslSettings holds SASL authentication settings.

type ScramSecretError 

type ScramSecretError struct {
	SecretArn    string `json:"secretArn"`
	ErrorCode    string `json:"errorCode,omitempty"`
	ErrorMessage string `json:"errorMessage,omitempty"`
}

ScramSecretError represents an error that occurred while associating or disassociating a SCRAM secret.

type ServerlessClientAuthentication 

type ServerlessClientAuthentication struct {
	Sasl *SaslSettings `json:"sasl,omitempty"`
}

ServerlessClientAuthentication holds authentication settings for a serverless cluster.

type ServerlessClusterInfo 

type ServerlessClusterInfo struct {
	ClientAuthentication *ServerlessClientAuthentication `json:"clientAuthentication,omitempty"`
	VpcConfigs           []ServerlessVpcConfig           `json:"vpcConfigs,omitempty"`
}

ServerlessClusterInfo holds serverless-specific cluster configuration.

type ServerlessVpcConfig 

type ServerlessVpcConfig struct {
	SubnetIDs        []string `json:"subnetIds,omitempty"`
	SecurityGroupIDs []string `json:"securityGroupIds,omitempty"`
}

ServerlessVpcConfig holds VPC configuration for a serverless cluster.

type StateInfo 

type StateInfo struct {
	Code    string `json:"code,omitempty"`
	Message string `json:"message,omitempty"`
}

StateInfo holds cluster state detail for error conditions.

type StorageBackend 

type StorageBackend interface {
	// Cluster operations
	CreateCluster(
		ctx context.Context,
		name, kafkaVersion string,
		numBrokers int32,
		brokerInfo BrokerNodeGroupInfo,
		clientAuth *ClientAuthentication,
		tags map[string]string,
	) (*Cluster, error)
	CreateServerlessCluster(
		ctx context.Context, name string, serverless *ServerlessClusterInfo, tags map[string]string,
	) (*Cluster, error)
	DescribeCluster(ctx context.Context, clusterArn string) (*Cluster, error)
	ListClusters(ctx context.Context) []*Cluster
	DeleteCluster(ctx context.Context, clusterArn string) error

	// Configuration operations
	CreateConfiguration(
		ctx context.Context,
		name, description string,
		kafkaVersions []string,
		serverProperties string,
	) (*Configuration, error)
	DescribeConfiguration(ctx context.Context, configArn string) (*Configuration, error)
	ListConfigurations(ctx context.Context) []*Configuration
	DeleteConfiguration(ctx context.Context, configArn string) error

	// Tag operations
	TagResource(ctx context.Context, resourceArn string, tags map[string]string) error
	UntagResource(ctx context.Context, resourceArn string, tagKeys []string) error
	GetTags(ctx context.Context, resourceArn string) (map[string]string, error)

	// SCRAM secret operations
	BatchAssociateScramSecret(
		ctx context.Context,
		clusterArn string,
		secretArnList []string,
	) ([]ScramSecretError, error)
	BatchDisassociateScramSecret(
		ctx context.Context, clusterArn string, secretArnList []string,
	) ([]ScramSecretError, error)

	// Replicator operations
	CreateReplicator(
		ctx context.Context,
		name, description, serviceExecutionRoleArn string,
		kafkaClusters []ClusterConfig,
		replicationInfoList []ReplicationInfoConfig,
		tags map[string]string,
	) (*Replicator, error)
	DeleteReplicator(ctx context.Context, replicatorArn string) error
	DescribeReplicator(ctx context.Context, replicatorArn string) (*Replicator, error)
	ListReplicators(ctx context.Context) []*Replicator
	UpdateReplicationInfo(
		ctx context.Context,
		replicatorArn, currentVersion, sourceKafkaClusterArn, targetKafkaClusterArn string,
		topicReplication *TopicReplicationConfig,
		consumerGroupReplication *ConsumerGroupReplicationConfig,
	) (*Replicator, error)

	// Topic operations
	CreateTopic(
		ctx context.Context,
		clusterArn, topicName string,
		replicationFactor, partitionCount int32,
		configs string,
	) (*Topic, error)
	DeleteTopic(ctx context.Context, clusterArn, topicName string) error
	DescribeTopic(ctx context.Context, clusterArn, topicName string) (*Topic, error)
	DescribeTopicPartitions(ctx context.Context, clusterArn, topicName string) ([]*TopicPartitionInfo, error)
	ListTopics(ctx context.Context, clusterArn, topicNameFilter string) ([]*Topic, error)
	UpdateTopic(
		ctx context.Context, clusterArn, topicName string, partitionCount int32, configs string,
	) (*Topic, error)

	// VPC connection operations
	CreateVpcConnection(
		ctx context.Context,
		targetClusterArn, vpcID, authentication string,
		clientSubnets, securityGroups []string,
		tags map[string]string,
	) (*VpcConnection, error)
	DeleteVpcConnection(ctx context.Context, vpcConnectionArn string) error
	DescribeVpcConnection(ctx context.Context, vpcConnectionArn string) (*VpcConnection, error)
	ListVpcConnections(ctx context.Context) []*VpcConnection
	ListClientVpcConnections(ctx context.Context, clusterArn string) ([]*VpcConnection, error)
	RejectClientVpcConnection(ctx context.Context, vpcConnectionArn string) error

	// Cluster policy operations
	DeleteClusterPolicy(ctx context.Context, clusterArn string) error
	GetClusterPolicy(ctx context.Context, clusterArn string) (string, error)
	PutClusterPolicy(ctx context.Context, clusterArn, policy string) error

	// Cluster operation operations
	DescribeClusterOperation(ctx context.Context, clusterOperationArn string) (*ClusterOperation, error)
	DescribeClusterOperationV2(ctx context.Context, clusterOperationArn string) (*ClusterOperation, error)
	ListClusterOperations(ctx context.Context, clusterArn string) ([]*ClusterOperation, error)
	ListClusterOperationsV2(ctx context.Context, clusterArn string) ([]*ClusterOperation, error)

	// Configuration revision operations
	DescribeConfigurationRevision(ctx context.Context, configArn string, revision int64) (*ConfigurationRevision, error)
	UpdateConfiguration(ctx context.Context, configArn, description, serverProperties string) (*Configuration, error)
	ListConfigurationRevisions(ctx context.Context, configArn string) ([]*ConfigurationRevision, error)

	// Broker / cluster update operations
	UpdateBrokerCount(ctx context.Context, clusterArn string, numBrokers int32) (*ClusterOperation, error)
	UpdateBrokerStorage(ctx context.Context, clusterArn string, volumeSize int32) (*ClusterOperation, error)
	UpdateBrokerType(ctx context.Context, clusterArn, instanceType string) (*ClusterOperation, error)
	UpdateClusterConfiguration(
		ctx context.Context,
		clusterArn, configArn string,
		revision int64,
	) (*ClusterOperation, error)
	UpdateClusterKafkaVersion(ctx context.Context, clusterArn, targetKafkaVersion string) (*ClusterOperation, error)
	UpdateConnectivity(
		ctx context.Context,
		clusterArn string,
		settings UpdateConnectivitySettings,
	) (*ClusterOperation, error)
	UpdateMonitoring(
		ctx context.Context,
		clusterArn string,
		settings UpdateMonitoringSettings,
	) (*ClusterOperation, error)
	UpdateRebalancing(ctx context.Context, clusterArn string) (*ClusterOperation, error)
	UpdateSecurity(ctx context.Context, clusterArn string, settings UpdateSecuritySettings) (*ClusterOperation, error)
	UpdateStorage(ctx context.Context, clusterArn string, settings UpdateStorageSettings) (*ClusterOperation, error)
	RebootBroker(ctx context.Context, clusterArn string, brokerIDs []string) (*ClusterOperation, error)

	// SCRAM secret list
	ListScramSecrets(ctx context.Context, clusterArn string) ([]string, error)

	// Node / version ops
	ListNodes(ctx context.Context, clusterArn string) ([]*BrokerNode, error)
	ListKafkaVersions(ctx context.Context) []*MSKVersion
	GetCompatibleKafkaVersions(ctx context.Context, clusterArn string) ([]*MSKVersion, error)

	// Lifecycle
	Reset()
	Region() string
	AccountID() string
	Snapshot(ctx context.Context) []byte
	Restore(ctx context.Context, data []byte) error
}

StorageBackend defines the interface for Kafka (MSK) backend implementations. All mutating methods must be safe for concurrent use.

Every operation takes a context.Context so the backend can resolve the caller's AWS region (for create/list operations) and route to the correct per-region store. Operations that target an existing resource ARN resolve their region from the ARN itself, falling back to the context region.

type StorageInfo 

type StorageInfo struct {
	EbsStorageInfo *EBSStorageInfo `json:"ebsStorageInfo,omitempty"`
}

StorageInfo holds broker storage config.

type TLSSettings 

type TLSSettings struct {
	CertificateAuthorityArnList []string `json:"certificateAuthorityArnList,omitempty"`
	Enabled                     bool     `json:"enabled"`
}

TLSSettings holds TLS authentication settings.

type Topic 

type Topic struct {
	ClusterArn        string `json:"clusterArn"`
	Configs           string `json:"configs,omitempty"`
	Status            string `json:"status"`
	TopicArn          string `json:"topicArn"`
	TopicName         string `json:"topicName"`
	PartitionCount    int32  `json:"partitionCount"`
	ReplicationFactor int32  `json:"replicationFactor"`
}

Topic represents an MSK topic on a cluster. ClusterArn is persisted (it is load-bearing for the primary key -- see topicKey/topicKeyFn -- and the topicsByCluster index, both of which are rebuilt from this field on Restore) but is NOT part of the real wire response; handlers must build a dedicated DTO for DescribeTopic/CreateTopic/UpdateTopic responses (see describeTopicOutputFrom in handler_topics.go) rather than marshaling a *Topic directly, or ClusterArn would leak into the API response.

type TopicInfo added in v1.2.0 

type TopicInfo struct {
	TopicArn              string `json:"topicArn"`
	TopicName             string `json:"topicName"`
	PartitionCount        int32  `json:"partitionCount"`
	ReplicationFactor     int32  `json:"replicationFactor"`
	OutOfSyncReplicaCount int32  `json:"outOfSyncReplicaCount"`
}

TopicInfo is the per-topic summary element returned by ListTopics, matching aws-sdk-go-v2/service/kafka/types.TopicInfo. It is a distinct (smaller) shape from Topic/DescribeTopicOutput: no configs/status, plus outOfSyncReplicaCount.

type TopicPartitionInfo added in v1.2.0 

type TopicPartitionInfo struct {
	Isr       []int32 `json:"isr"`
	Replicas  []int32 `json:"replicas"`
	Partition int32   `json:"partition"`
	Leader    int32   `json:"leader"`
}

TopicPartitionInfo is the per-partition element returned by DescribeTopicPartitions, matching types.TopicPartitionInfo.

type TopicReplicationConfig added in v1.2.0 

type TopicReplicationConfig struct {
	StartingPositionType            string   `json:"startingPositionType,omitempty"`
	TopicNameConfigurationType      string   `json:"topicNameConfigurationType,omitempty"`
	TopicsToExclude                 []string `json:"topicsToExclude,omitempty"`
	TopicsToReplicate               []string `json:"topicsToReplicate,omitempty"`
	CopyAccessControlListsForTopics bool     `json:"copyAccessControlListsForTopics"`
	CopyTopicConfigurations         bool     `json:"copyTopicConfigurations"`
	DetectAndCopyNewTopics          bool     `json:"detectAndCopyNewTopics"`
}

TopicReplicationConfig mirrors types.TopicReplication / types.TopicReplicationUpdate: the topic-replication half of a ReplicationInfo entry.

type UnauthenticatedSettings 

type UnauthenticatedSettings struct {
	Enabled bool `json:"enabled"`
}

UnauthenticatedSettings holds unauthenticated access settings.

type UpdateConnectivitySettings 

type UpdateConnectivitySettings struct {
	ConnectivityInfo *ConnectivityInfo
}

UpdateConnectivitySettings is the payload for UpdateConnectivity.

type UpdateMonitoringSettings 

type UpdateMonitoringSettings struct {
	OpenMonitoring     *OpenMonitoring
	LoggingInfo        *LoggingInfo
	EnhancedMonitoring string
}

UpdateMonitoringSettings is the payload for UpdateMonitoring.

type UpdateSecuritySettings 

type UpdateSecuritySettings struct {
	ClientAuthentication *ClientAuthentication
	EncryptionInfo       *EncryptionInfo
}

UpdateSecuritySettings is the payload for UpdateSecurity.

type UpdateStorageSettings 

type UpdateStorageSettings struct {
	ProvisionedThroughput *ProvisionedThroughput
	StorageMode           string
	VolumeSizeGB          int32
}

UpdateStorageSettings is the payload for UpdateStorage.

type VpcConnection 

type VpcConnection struct {
	Tags             map[string]string `json:"-"`
	VpcConnectionArn string            `json:"vpcConnectionArn"`
	TargetClusterArn string            `json:"targetClusterArn"`
	VpcID            string            `json:"vpcId"`
	Authentication   string            `json:"authentication,omitempty"`
	State            string            `json:"state"`
	CreationTime     string            `json:"creationTime,omitempty"`
	SubnetIDs        []string          `json:"subnetIds,omitempty"`
	SecurityGroupIDs []string          `json:"securityGroupIds,omitempty"`
}

VpcConnection represents an MSK VPC connection.

type VpcConnectivity 

type VpcConnectivity struct {
	ClientAuthentication *VpcConnectivityClientAuthentication `json:"clientAuthentication,omitempty"`
}

VpcConnectivity holds VPC connectivity configuration.

type VpcConnectivityClientAuthentication 

type VpcConnectivityClientAuthentication struct {
	Sasl *VpcConnectivitySasl `json:"sasl,omitempty"`
	TLS  *VpcConnectivityTLS  `json:"tls,omitempty"`
}

VpcConnectivityClientAuthentication holds authentication settings for VPC connectivity.

type VpcConnectivitySasl 

type VpcConnectivitySasl struct {
	Iam   *VpcConnectivitySaslIam   `json:"iam,omitempty"`
	Scram *VpcConnectivitySaslScram `json:"scram,omitempty"`
}

VpcConnectivitySasl holds SASL settings for VPC connectivity.

type VpcConnectivitySaslIam 

type VpcConnectivitySaslIam struct {
	Enabled bool `json:"enabled"`
}

VpcConnectivitySaslIam holds IAM authentication settings for VPC connectivity.

type VpcConnectivitySaslScram 

type VpcConnectivitySaslScram struct {
	Enabled bool `json:"enabled"`
}

VpcConnectivitySaslScram holds SCRAM authentication settings for VPC connectivity.

type VpcConnectivityTLS 

type VpcConnectivityTLS struct {
	Enabled bool `json:"enabled"`
}

VpcConnectivityTLS holds TLS settings for VPC connectivity.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.