kafka

package
v1.5.0 Latest Latest
Warning

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

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

README

Managed Streaming for Kafka

Parity grade: A · SDK aws-sdk-go-v2/service/kafka@v1.57.2 · last audited 2026-08-05 (fcb3fbbb9f46c11d4cf4034410f5ec80e7f16f63)

Coverage

Metric Value
PARITY entries audited 64 (63 ok, 1 partial)
Feature families 13 (13 ok)
Known gaps 3
Deferred items 0
Resource leaks clean
Known gaps
  • "Channel Create/Update/Delete are immediate (no CREATING/UPDATING/DELETING polling window) -- same documented simplification as Topic.Status (see below): the real API exposes a ClusterOperationArn/polling protocol this in-memory emulator has no async execution to model, so Channel.Status goes straight to ACTIVE and ClusterOperationArn is populated only on the mutating call's own response, never on the persisted record (matching what a real client would observe once the real async operation has already completed by the time it calls Describe)."
  • "CreateChannel does not restrict channel creation to MSK Express clusters, even though CreateChannel's doc comment says a channel streams from 'an Amazon MSK Express cluster topic'. gopherstack's Cluster model has no Express-vs-standard-broker-type distinction anywhere else in this service, and the SDK's client-side validators.go does not enforce it either (it can only be a server-side rule), so modeling this specific restriction here would mean inventing a cluster-type check found nowhere else in the codebase rather than verifying one against the SDK."
  • "CreateChannel does not verify that TopicConfigurationList[].TopicArn references a topic that actually exists in this backend. The real service's behavior here is unverifiable from the client SDK alone (no client-side check exists in validators.go), so enforcing an invented rule risks fabricating unproven behavior; the ARN is accepted, stored, and echoed back verbatim instead." # All 5 gaps from the 2026-07-12 audit (topic field names, DescribeTopicPartitions # shape, UpdateReplicationInfo shape, CreateReplicator missing topology fields, # Cluster.CurrentVersion never advancing) are closed -- see the op/family notes # above for exactly what changed and where. Two NEW real wire bugs were found and # fixed while closing out the deferred items below (GetBootstrapBrokers field # names, ListClientVpcConnections envelope+shape) plus a missing-required-field # gap on CreateVpcConnection/DescribeVpcConnection (clientSubnets/securityGroups). # # Documented simplifications (not wire-shape gaps -- these are internal-model # choices that do not diverge from any real MSK response field or type): # - Topic.Status is always ACTIVE immediately on Create/Update; real MSK's # TopicState enum also has CREATING/UPDATING/DELETING but topic creation # exposes no polling protocol the way cluster creation does, so there is no # externally observable "stuck CREATING" behavior to get wrong. # - DescribeTopicPartitions' Isr is always == Replicas (fully in-sync); this # in-memory emulator has no real per-broker replication lag to diverge from. # - ClientVpcConnection.Owner is populated from the backend's own AccountID as # a best-effort placeholder; gopherstack has no cross-account VPC-connection # ownership model to draw a different value from.

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"
)
View Source 
const (
	ChannelDestinationTypeS3      = "S3"
	ChannelDestinationTypeIceberg = "ICEBERG"
)

ChannelDestinationType* mirror types.ChannelDestinationType.

View Source 
const (
	ChannelStatusActive     = "ACTIVE"
	ChannelStatusCreating   = "CREATING"
	ChannelStatusUpdating   = "UPDATING"
	ChannelStatusDeleting   = "DELETING"
	ChannelStatusFailed     = "FAILED"
	ChannelStatusSuspending = "SUSPENDING"
	ChannelStatusSuspended  = "SUSPENDED"
)

ChannelStatus* mirror types.ChannelStatus.

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 Catalog added in v1.3.1 

type Catalog struct {
	CatalogArn        string `json:"catalogArn,omitempty"`
	WarehouseLocation string `json:"warehouseLocation,omitempty"`
}

Catalog mirrors types.Catalog.

type Channel added in v1.3.1 

type Channel struct {
	Tags                            map[string]string                `json:"tags,omitempty"`
	EncryptionConfiguration         *ChannelEncryptionConfiguration  `json:"encryptionConfiguration,omitempty"`
	IcebergDestinationConfiguration *IcebergDestinationConfiguration `json:"icebergDestinationConfiguration,omitempty"`
	LoggingInfo                     *ChannelLoggingInfo              `json:"loggingInfo,omitempty"`
	S3DestinationConfiguration      *S3DestinationConfiguration      `json:"s3DestinationConfiguration,omitempty"`
	StateInfo                       *ChannelStateInfo                `json:"stateInfo,omitempty"`
	ClusterArn                      string                           `json:"clusterArn"`
	ChannelArn                      string                           `json:"channelArn"`
	ChannelName                     string                           `json:"channelName"`
	ClusterOperationArn             string                           `json:"clusterOperationArn,omitempty"`
	CreationTime                    string                           `json:"creationTime,omitempty"`
	DestinationType                 string                           `json:"destinationType"`
	Status                          string                           `json:"status"`
	TopicConfigurationList          []TopicConfiguration             `json:"topicConfigurationList,omitempty"`
}

Channel represents an MSK channel. ClusterArn is persisted for the ClusterArn-scope check and the channelsByCluster index but is NOT part of the real wire response — handlers build a dedicated DTO (see describeChannelOutputFrom), same pattern as Topic.

Unlike Cluster/Configuration/Replicator/VpcConnection, Tags uses a normal JSON tag, not json:"-": DescribeChannelOutput's wire shape includes "tags" directly (deserializers.go: awsRestjson1_deserializeOpDocumentDescribeChannelOutput), so it needs no fixNilTags special case on Snapshot/Restore (persistence.go).

type ChannelEncryptionConfiguration added in v1.3.1 

type ChannelEncryptionConfiguration struct {
	KmsKeyArn string `json:"kmsKeyArn,omitempty"`
}

ChannelEncryptionConfiguration mirrors types.EncryptionConfiguration: the channel-level KMS setting (distinct from Cluster.EncryptionInfo).

type ChannelLoggingInfo added in v1.3.1 

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

ChannelLoggingInfo mirrors types.ChannelLoggingInfo. It reuses the existing CloudWatchLogs/Firehose/S3Logs types, whose field names/shapes already match types.CloudWatchLogs/types.Firehose/types.S3 exactly.

type ChannelStateInfo added in v1.3.1 

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

ChannelStateInfo mirrors types.ChannelStateInfo: additional context for a channel in FAILED state.

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"`
	Rebalancing          *Rebalancing           `json:"rebalancing,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:"operationArn"`
	ClusterArn          string              `json:"clusterArn"`
	OperationType       string              `json:"operationType"`
	OperationState      string              `json:"operationState"`
}

ClusterOperation represents an MSK cluster operation.

type CompatibleKafkaVersion added in v1.5.0 

type CompatibleKafkaVersion struct {
	SourceVersion  string   `json:"sourceVersion"`
	TargetVersions []string `json:"targetVersions"`
}

CompatibleKafkaVersion groups the version a cluster is upgrading FROM with the versions it can upgrade TO, matching aws-sdk-go-v2 kafka's types.CompatibleKafkaVersion (types/types.go:576; deserializers.go:15252 keys "sourceVersion"/"targetVersions").

type Configuration 

type Configuration struct {
	Tags             map[string]string      `json:"-"`
	LatestRevision   *ConfigurationRevision `json:"latestRevision"`
	Arn              string                 `json:"arn"`
	Name             string                 `json:"name"`
	Description      string                 `json:"description"`
	ServerProperties string                 `json:"serverProperties"`
	CreationTime     string                 `json:"creationTime"`
	State            string                 `json:"state"`
	KafkaVersions    []string               `json:"kafkaVersions"`
}

Configuration represents an MSK configuration. CreationTime/LatestRevision/ State/Description are all real required members of the SDK's types.Configuration (kafka@v1.57.2 types/types.go) -- ListConfigurations marshals this type directly as its list-item shape, so every field here must match that contract, not just DescribeConfiguration/CreateConfiguration's own (unrequired) op-level shapes.

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"`
	ServerProperties string `json:"serverProperties,omitempty"`
	CreationTime     string `json:"creationTime"`
	Revision         int64  `json:"revision"`
}

ConfigurationRevision represents a revision of an MSK configuration. CreationTime is a real required member of types.ConfigurationRevision -- ListConfigurationRevisions marshals this type directly as its list-item shape (deserializers.go's awsRestjson1_deserializeDocumentConfigurationRevision switches on "creationTime").

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 DeadLetterQueueS3 added in v1.3.1 

type DeadLetterQueueS3 struct {
	BucketArn           string `json:"bucketArn,omitempty"`
	ErrorOutputPrefix   string `json:"errorOutputPrefix,omitempty"`
	ExpectedBucketOwner string `json:"expectedBucketOwner,omitempty"`
}

DeadLetterQueueS3 mirrors types.DeadLetterQueueS3.

type DestinationTable added in v1.3.1 

type DestinationTable struct {
	PartitionSpec           *PartitionSpec `json:"partitionSpec,omitempty"`
	DestinationDatabaseName string         `json:"destinationDatabaseName,omitempty"`
	DestinationTableName    string         `json:"destinationTableName,omitempty"`
}

DestinationTable mirrors types.DestinationTable.

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 IcebergDestinationConfiguration added in v1.3.1 

type IcebergDestinationConfiguration struct {
	Catalog                 *Catalog           `json:"catalog,omitempty"`
	DeadLetterQueueS3       *DeadLetterQueueS3 `json:"deadLetterQueueS3,omitempty"`
	SchemaEvolution         *SchemaEvolution   `json:"schemaEvolution,omitempty"`
	TableCreation           *TableCreation     `json:"tableCreation,omitempty"`
	CompressionType         string             `json:"compressionType,omitempty"`
	ServiceExecutionRoleArn string             `json:"serviceExecutionRoleArn,omitempty"`
	DestinationTableList    []DestinationTable `json:"destinationTableList,omitempty"`
	DataFreshnessInSeconds  int32              `json:"dataFreshnessInSeconds,omitempty"`
	AppendOnly              bool               `json:"appendOnly"`
}

IcebergDestinationConfiguration mirrors types.IcebergDestinationConfiguration.

type IcebergDestinationUpdate added in v1.3.1 

type IcebergDestinationUpdate struct {
	DataFreshnessInSeconds int32 `json:"dataFreshnessInSeconds"`
}

IcebergDestinationUpdate mirrors types.IcebergDestinationUpdate.

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) CreateChannel added in v1.3.1 

func (b *InMemoryBackend) CreateChannel(
	ctx context.Context,
	clusterArn, channelName string,
	topicConfigurationList []TopicConfiguration,
	encryptionConfiguration *ChannelEncryptionConfiguration,
	icebergDestinationConfiguration *IcebergDestinationConfiguration,
	s3DestinationConfiguration *S3DestinationConfiguration,
	loggingInfo *ChannelLoggingInfo,
	tags map[string]string,
) (*Channel, error)

CreateChannel creates a channel on an MSK cluster. Real MSK creation is async (Status starts CREATING); this emulator makes it ACTIVE immediately, matching CreateTopic's simplification. ClusterOperationArn is returned but not persisted — real DescribeChannel/ListChannels would show it empty too once Status is ACTIVE (it's only set while CREATING/UPDATING/DELETING, per types.go's DescribeChannelOutput doc comment).

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,
	logDelivery *LogDelivery,
) (*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) DeleteChannel added in v1.3.1 

func (b *InMemoryBackend) DeleteChannel(ctx context.Context, clusterArn, channelArn string) (*Channel, error)

DeleteChannel deletes a channel from an MSK cluster. Real MSK deletion is async (DELETING); this emulator removes it immediately.

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) DescribeChannel added in v1.3.1 

func (b *InMemoryBackend) DescribeChannel(_ context.Context, clusterArn, channelArn string) (*Channel, error)

DescribeChannel retrieves a channel by cluster ARN and channel ARN. A channelArn that exists but belongs to a different cluster is reported as not found, matching real MSK's cluster-scoped Channel resource model.

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,
) ([]*CompatibleKafkaVersion, error)

GetCompatibleKafkaVersions returns the versions the cluster's current Kafka version can upgrade to, grouped by that current (source) 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, VPC connection, or channel by ARN.

func (*InMemoryBackend) ListChannels added in v1.3.1 

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

ListChannels returns channels for a cluster sorted by channel name, optionally filtered by topicNameFilter. Unlike ListTopics' prefix-match topicNameFilter, ListChannels' doc comment says "matches the specified value" with no "starting with" qualifier, so this is an exact match.

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, VPC connection, or channel by ARN.

func (*InMemoryBackend) UntagResource 

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

UntagResource removes tags from a cluster, configuration, replicator, VPC connection, or channel 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) UpdateChannel added in v1.3.1 

func (b *InMemoryBackend) UpdateChannel(
	ctx context.Context,
	clusterArn, channelArn string,
	icebergDestinationUpdate *IcebergDestinationUpdate,
	s3DestinationUpdate *S3DestinationUpdate,
) (*Channel, error)

UpdateChannel updates the destination-freshness setting of an existing channel. The destination type cannot be changed (api_op_UpdateChannel.go doc comment); enforced here server-side since neither update field is marked required in validators.go.

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, status string,
) (*ClusterOperation, error)

UpdateRebalancing updates a cluster's intelligent rebalancing status, persisting the new Rebalancing.Status and recording an operation whose source/target reflect the before/after state (mirroring UpdateMonitoring).

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 LogDelivery added in v1.3.1 

type LogDelivery struct {
	ReplicatorLogDelivery *ReplicatorLogDelivery `json:"replicatorLogDelivery,omitempty"`
}

LogDelivery configures log delivery for a Replicator. Field-diffed against kafka@v1.57.2 types.go's LogDelivery/ReplicatorLogDelivery: CreateReplicator accepted this on the request but previously discarded it entirely (not stored, not echoed by DescribeReplicator).

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"`
	Rebalancing          *Rebalancing          `json:"rebalancing,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 PartitionSource added in v1.3.1 

type PartitionSource struct {
	SourceName string `json:"sourceName,omitempty"`
}

PartitionSource mirrors types.PartitionSource.

type PartitionSpec added in v1.3.1 

type PartitionSpec struct {
	PartitionStrategy string            `json:"partitionStrategy,omitempty"`
	SourceList        []PartitionSource `json:"sourceList,omitempty"`
}

PartitionSpec mirrors types.PartitionSpec.

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 Rebalancing added in v1.3.1 

type Rebalancing struct {
	Status string `json:"status,omitempty"`
}

Rebalancing describes the intelligent rebalancing configuration of an MSK Provisioned cluster with Express brokers (aws-sdk-go-v2/service/kafka's types.Rebalancing; wire key "status").

type RecordConverter added in v1.3.1 

type RecordConverter struct {
	ValueConverter string `json:"valueConverter,omitempty"`
}

RecordConverter mirrors types.RecordConverter.

type RecordSchema added in v1.3.1 

type RecordSchema struct {
	GsrArn string `json:"gsrArn,omitempty"`
}

RecordSchema mirrors types.RecordSchema.

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:"-"`
	LogDelivery             *LogDelivery            `json:"logDelivery,omitempty"`
	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 ReplicatorLogDelivery added in v1.3.1 

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

ReplicatorLogDelivery configures where a replicator's logs are delivered. CloudWatchLogs/Firehose/S3Logs are reused as-is: their wire field names (enabled/logGroup, enabled/deliveryStream, enabled/bucket/prefix) are identical to the real ReplicatorCloudWatchLogs/ReplicatorFirehose/ReplicatorS3 types, confirmed against types.go.

type S3DestinationConfiguration added in v1.3.1 

type S3DestinationConfiguration struct {
	DeadLetterQueueS3       *DeadLetterQueueS3 `json:"deadLetterQueueS3,omitempty"`
	Storage                 *S3Storage         `json:"storage,omitempty"`
	ServiceExecutionRoleArn string             `json:"serviceExecutionRoleArn,omitempty"`
	DataFreshnessInSeconds  int32              `json:"dataFreshnessInSeconds,omitempty"`
}

S3DestinationConfiguration mirrors types.S3DestinationConfiguration.

type S3DestinationUpdate added in v1.3.1 

type S3DestinationUpdate struct {
	DataFreshnessInSeconds int32 `json:"dataFreshnessInSeconds"`
}

S3DestinationUpdate mirrors types.S3DestinationUpdate.

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 S3Storage added in v1.3.1 

type S3Storage struct {
	BucketArn           string `json:"bucketArn,omitempty"`
	CompressionType     string `json:"compressionType,omitempty"`
	StorageClass        string `json:"storageClass,omitempty"`
	ExpectedBucketOwner string `json:"expectedBucketOwner,omitempty"`
	OutputKeyTemplate   string `json:"outputKeyTemplate,omitempty"`
	OutputPrefix        string `json:"outputPrefix,omitempty"`
}

S3Storage mirrors types.S3Storage.

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 SchemaEvolution added in v1.3.1 

type SchemaEvolution struct {
	EnableSchemaEvolution bool `json:"enableSchemaEvolution"`
}

SchemaEvolution mirrors types.SchemaEvolution.

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,
		logDelivery *LogDelivery,
	) (*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, status 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) ([]*CompatibleKafkaVersion, error)

	// Channel operations
	CreateChannel(
		ctx context.Context,
		clusterArn, channelName string,
		topicConfigurationList []TopicConfiguration,
		encryptionConfiguration *ChannelEncryptionConfiguration,
		icebergDestinationConfiguration *IcebergDestinationConfiguration,
		s3DestinationConfiguration *S3DestinationConfiguration,
		loggingInfo *ChannelLoggingInfo,
		tags map[string]string,
	) (*Channel, error)
	DeleteChannel(ctx context.Context, clusterArn, channelArn string) (*Channel, error)
	DescribeChannel(ctx context.Context, clusterArn, channelArn string) (*Channel, error)
	ListChannels(ctx context.Context, clusterArn, topicNameFilter string) ([]*Channel, error)
	UpdateChannel(
		ctx context.Context,
		clusterArn, channelArn string,
		icebergDestinationUpdate *IcebergDestinationUpdate,
		s3DestinationUpdate *S3DestinationUpdate,
	) (*Channel, 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 TableCreation added in v1.3.1 

type TableCreation struct {
	EnableTableCreation bool `json:"enableTableCreation"`
}

TableCreation mirrors types.TableCreation.

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 TopicConfiguration added in v1.3.1 

type TopicConfiguration struct {
	RecordConverter *RecordConverter `json:"recordConverter,omitempty"`
	RecordSchema    *RecordSchema    `json:"recordSchema,omitempty"`
	TopicArn        string           `json:"topicArn,omitempty"`
}

TopicConfiguration mirrors types.TopicConfiguration: the Kafka topic that feeds a channel, plus how record values are deserialized.

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.