kafkaconnect

package
v1.7.0 Latest Latest
Warning

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

 Go to latest Published: Oct 6, 2026 License: MIT Imports: 32 Imported by: 0

README

Kafkaconnect

Parity grade: A · SDK aws-sdk-go-v2/service/kafkaconnect@v1.39.1 · last audited 2026-10-05 (5bb0d02ee)

Coverage

Metric Value
PARITY entries audited 19 (19 ok)
Feature families 5 (5 ok)
Known gaps none
Structural gaps (can't be emulated) 3
Deferred items 0
Resource leaks clean
Structural gaps

These do not block an A grade — no implementation could produce real data here because the underlying data source cannot exist in an emulator.

  • No Kafka Connect runtime: connector tasks never run, there is no worker fleet or autoscaling metric source, and MSK bootstrap-broker / VPC reachability, connector.class resolution and plugin-archive validity are never exercised. AWS surfaces those failures asynchronously as FAILED states whose StateDescription code vocabulary is not published, so connectors here never reach FAILED.
  • A custom plugin whose S3 object is absent ends CREATE_FAILED with a message-only StateDescription (the code vocabulary is unpublished); archive contents are not validated as a real ZIP/JAR.
  • UpdateCustomPlugin / UpdateWorkerConfiguration do not exist in the pinned SDK (v1.39.1), so plugin and worker-configuration revisions stay at 1.

More

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	// ErrConnectorNotFound is returned when a connector ARN does not resolve.
	ErrConnectorNotFound = awserr.New("connector not found", awserr.ErrNotFound)
	// ErrConnectorNameInUse is returned when a connector name is already in use.
	ErrConnectorNameInUse = awserr.New("a connector with this name already exists", awserr.ErrAlreadyExists)
	// ErrCustomPluginNotFound is returned when a custom plugin ARN does not resolve.
	ErrCustomPluginNotFound = awserr.New("custom plugin not found", awserr.ErrNotFound)
	// ErrCustomPluginNameInUse is returned when a custom plugin name is already in use.
	ErrCustomPluginNameInUse = awserr.New("a custom plugin with this name already exists", awserr.ErrAlreadyExists)
	// ErrWorkerConfigNotFound is returned when a worker configuration ARN does not resolve.
	ErrWorkerConfigNotFound = awserr.New("worker configuration not found", awserr.ErrNotFound)
	// ErrWorkerConfigNameInUse is returned when a worker configuration name is already in use.
	ErrWorkerConfigNameInUse = awserr.New(
		"a worker configuration with this name already exists",
		awserr.ErrAlreadyExists,
	)
	// ErrConnectorOperationNotFound is returned when a connector operation ARN does not resolve.
	ErrConnectorOperationNotFound = awserr.New("connector operation not found", awserr.ErrNotFound)
	// ErrResourceNotFound is returned by the generic tag operations when resourceArn resolves to nothing.
	ErrResourceNotFound = awserr.New("resource not found", awserr.ErrNotFound)
	// ErrVersionMismatch is returned when currentVersion does not match a connector's actual version.
	ErrVersionMismatch = awserr.New(
		"the current version specified does not match the connector's actual current version",
		awserr.ErrConflict,
	)
	// ErrConnectorDeleting is returned when mutating a connector that is already DELETING.
	ErrConnectorDeleting = awserr.New("the connector is being deleted", awserr.ErrConflict)
	// ErrValidation is returned when request input fails validation.
	ErrValidation = awserr.New("invalid request", awserr.ErrInvalidParameter)
)
View Source 
var ErrNoSnapshot = errors.New("backend does not support restore")

ErrNoSnapshot is returned when a backend does not support snapshot/restore.

Functions

This section is empty.

Types

type ApacheKafkaCluster 

type ApacheKafkaCluster struct {
	BootstrapServers string
	Vpc              Vpc
}

ApacheKafkaCluster mirrors types.ApacheKafkaClusterDescription.

type AutoScaling 

type AutoScaling struct {
	MinWorkerCount          int32
	MaxWorkerCount          int32
	McuCount                int32
	MaxAutoscalingTaskCount int32
	ScaleInCPUPercent       int32
	ScaleOutCPUPercent      int32
}

AutoScaling mirrors types.AutoScalingDescription.

type Capacity 

type Capacity struct {
	AutoScaling *AutoScaling
	Provisioned *ProvisionedCapacity
}

Capacity mirrors types.CapacityDescription: exactly one of the two is set.

type CloudWatchLogsDelivery 

type CloudWatchLogsDelivery struct {
	LogGroup string
	Enabled  bool
}

CloudWatchLogsDelivery mirrors types.CloudWatchLogsLogDeliveryDescription.

type Connector 

type Connector struct {
	CreationTime                     time.Time
	PendingUntil                     time.Time
	Capacity                         Capacity
	WorkerConfiguration              *WorkerConfigRef
	WorkerLogDelivery                *WorkerLogDelivery
	ConnectorConfiguration           map[string]string
	Tags                             map[string]string
	State                            string
	Description                      string
	ARN                              string
	CurrentVersion                   string
	KafkaConnectVersion              string
	KafkaClusterClientAuthentication string
	KafkaClusterEncryptionInTransit  string
	ServiceExecutionRoleArn          string
	NetworkType                      string
	Name                             string
	ApacheKafkaCluster               ApacheKafkaCluster
	Plugins                          []PluginRef
}

Connector is the persisted representation of an MSK Connect connector.

type ConnectorOperation 

type ConnectorOperation struct {
	CreationTime                 time.Time
	EndTime                      time.Time
	OriginConnectorConfiguration map[string]string
	TargetConnectorConfiguration map[string]string
	OriginCapacity               *Capacity
	TargetCapacity               *Capacity
	ARN                          string
	ConnectorArn                 string
	State                        string
	Type                         string
	Steps                        []ConnectorOperationStep
}

ConnectorOperation is the persisted representation of an UpdateConnector operation, returned by DescribeConnectorOperation/ListConnectorOperations.

type ConnectorOperationStep 

type ConnectorOperationStep struct {
	StepType  string
	StepState string
}

ConnectorOperationStep mirrors types.ConnectorOperationStep.

type ConnectorSpec 

type ConnectorSpec struct {
	Capacity                         Capacity
	ConnectorConfiguration           map[string]string
	Tags                             map[string]string
	WorkerConfiguration              *WorkerConfigRef
	WorkerLogDelivery                *WorkerLogDelivery
	Description                      string
	Name                             string
	KafkaConnectVersion              string
	KafkaClusterClientAuthentication string
	KafkaClusterEncryptionInTransit  string
	ServiceExecutionRoleArn          string
	NetworkType                      string
	ApacheKafkaCluster               ApacheKafkaCluster
	Plugins                          []PluginRef
}

ConnectorSpec carries every CreateConnector input field except the account/region needed to mint the ARN.

type ConnectorUpdate 

type ConnectorUpdate struct {
	Capacity               *Capacity
	ConnectorConfiguration map[string]string
}

ConnectorUpdate carries UpdateConnector's mutually exclusive target fields; exactly one of Capacity or ConnectorConfiguration is non-nil.

type CustomPlugin 

type CustomPlugin struct {
	CreationTime   time.Time
	PendingUntil   time.Time
	Tags           map[string]string
	Name           string
	ARN            string
	Description    string
	State          string
	FailureMessage string
	ContentType    string
	BucketArn      string
	FileKey        string
	ObjectVersion  string
	FileMD5        string
	FileSizeBytes  int64
	Revision       int64
}

CustomPlugin is the persisted representation of an MSK Connect custom plugin.

type FirehoseDelivery 

type FirehoseDelivery struct {
	DeliveryStream string
	Enabled        bool
}

FirehoseDelivery mirrors types.FirehoseLogDeliveryDescription.

type Handler 

type Handler struct {
	Backend StorageBackend

	AccountID     string
	DefaultRegion string
	// contains filtered or unexported fields
}

Handler is the HTTP handler for the MSK Connect REST API.

func NewHandler 

func NewHandler(backend StorageBackend) *Handler

NewHandler creates a new MSK Connect handler.

func (*Handler) BackendFor 

func (h *Handler) BackendFor(region string) StorageBackend

BackendFor returns the backend serving region: the home backend, or the sibling built on first use.

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 handles.

func (*Handler) ChaosServiceName 

func (h *Handler) ChaosServiceName() string

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

func (*Handler) EnableRegions 

func (h *Handler) EnableRegions()

EnableRegions makes h serve every other region through lazily built per-region siblings.

func (*Handler) ExtractOperation 

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

ExtractOperation extracts the MSK Connect operation name from the request.

func (*Handler) ExtractResource 

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

ExtractResource extracts the resource ARN from the request path.

func (*Handler) GetSupportedOperations 

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

GetSupportedOperations returns the list of supported operations.

func (*Handler) Handler 

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

Handler serves each request from the handler of its region.

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) RegionBackends 

func (h *Handler) RegionBackends() []StorageBackend

RegionBackends returns the home backend followed by every sibling built so far.

func (*Handler) RegionHandler 

func (h *Handler) RegionHandler(region string) *Handler

RegionHandler returns the handler serving region: h itself for the home region, else its sibling.

func (*Handler) Reset 

func (h *Handler) Reset()

Reset clears every region.

func (*Handler) Restore 

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

Restore implements persistence.Persistable.

func (*Handler) RouteMatcher 

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

RouteMatcher returns a function that matches MSK Connect REST API requests.

func (*Handler) Snapshot 

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

Snapshot implements persistence.Persistable; other regions ride in an additive "regions" key.

type InMemoryBackend 

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

InMemoryBackend is the in-memory implementation of StorageBackend.

func NewInMemoryBackend 

func NewInMemoryBackend() *InMemoryBackend

NewInMemoryBackend creates a new in-memory MSK Connect backend.

func (*InMemoryBackend) CreateConnector 

func (b *InMemoryBackend) CreateConnector(accountID, region string, spec ConnectorSpec) (*Connector, error)

CreateConnector creates a connector in CREATING; it settles to RUNNING after provisionDelay.

func (*InMemoryBackend) CreateCustomPlugin 

func (b *InMemoryBackend) CreateCustomPlugin(
	accountID, region, name, description, contentType, bucketArn, fileKey, objectVersion string,
	tags map[string]string,
) (*CustomPlugin, error)

CreateCustomPlugin creates a plugin in CREATING that settles to ACTIVE, or CREATE_FAILED without its S3 object; FileMd5/FileSize come from S3 when wired, else are derived.

func (*InMemoryBackend) CreateWorkerConfiguration 

func (b *InMemoryBackend) CreateWorkerConfiguration(
	accountID, region, name, description, propertiesFileContent string,
	tags map[string]string,
) (*WorkerConfiguration, error)

CreateWorkerConfiguration creates a worker configuration, always at revision 1 and ACTIVE immediately -- UpdateWorkerConfiguration (which would create later revisions) is not part of the AWS API surface this backend implements; see PARITY.md.

func (*InMemoryBackend) DeleteConnector 

func (b *InMemoryBackend) DeleteConnector(connectorArn, currentVersion string) (*Connector, error)

DeleteConnector marks a connector DELETING under optimistic lock (currentVersion, when supplied); it is removed once deletionDelay elapses.

func (*InMemoryBackend) DeleteCustomPlugin 

func (b *InMemoryBackend) DeleteCustomPlugin(customPluginArn string) (*CustomPlugin, error)

DeleteCustomPlugin marks a custom plugin DELETING; it is removed once deletionDelay elapses.

func (*InMemoryBackend) DeleteWorkerConfiguration 

func (b *InMemoryBackend) DeleteWorkerConfiguration(workerConfigurationArn string) (*WorkerConfiguration, error)

DeleteWorkerConfiguration marks a worker configuration DELETING; it is removed once deletionDelay elapses.

func (*InMemoryBackend) DescribeConnector 

func (b *InMemoryBackend) DescribeConnector(connectorArn string) (*Connector, error)

DescribeConnector returns the current information about a connector.

func (*InMemoryBackend) DescribeConnectorOperation 

func (b *InMemoryBackend) DescribeConnectorOperation(operationArn string) (*ConnectorOperation, error)

DescribeConnectorOperation returns the details of a single connector operation.

func (*InMemoryBackend) DescribeCustomPlugin 

func (b *InMemoryBackend) DescribeCustomPlugin(customPluginArn string) (*CustomPlugin, error)

DescribeCustomPlugin returns the current information about a custom plugin.

func (*InMemoryBackend) DescribeWorkerConfiguration 

func (b *InMemoryBackend) DescribeWorkerConfiguration(workerConfigurationArn string) (*WorkerConfiguration, error)

DescribeWorkerConfiguration returns the current information about a worker configuration.

func (*InMemoryBackend) ListConnectorOperations 

func (b *InMemoryBackend) ListConnectorOperations(
	connectorArn, nextToken string,
	maxResults int,
) ([]*ConnectorOperation, string, error)

ListConnectorOperations returns operations for a connector, paginated by nextToken/maxResults.

func (*InMemoryBackend) ListConnectors 

func (b *InMemoryBackend) ListConnectors(namePrefix, nextToken string, maxResults int) ([]*Connector, string, error)

ListConnectors returns connectors matching namePrefix, paginated by nextToken/maxResults.

func (*InMemoryBackend) ListCustomPlugins 

func (b *InMemoryBackend) ListCustomPlugins(
	namePrefix, nextToken string,
	maxResults int,
) ([]*CustomPlugin, string, error)

ListCustomPlugins returns custom plugins matching namePrefix, paginated by nextToken/maxResults.

func (*InMemoryBackend) ListTagsForResource 

func (b *InMemoryBackend) ListTagsForResource(resourceArn string) (map[string]string, error)

ListTagsForResource returns all tags on a connector, custom plugin, or worker configuration by ARN.

func (*InMemoryBackend) ListWorkerConfigurations 

func (b *InMemoryBackend) ListWorkerConfigurations(
	namePrefix, nextToken string,
	maxResults int,
) ([]*WorkerConfiguration, string, error)

ListWorkerConfigurations returns worker configurations matching namePrefix, paginated by nextToken/maxResults.

func (*InMemoryBackend) Reset 

func (b *InMemoryBackend) Reset()

Reset clears all backend state.

func (*InMemoryBackend) RestartConnector 

func (b *InMemoryBackend) RestartConnector(connectorArn string, _ bool) (*Connector, *ConnectorOperation, error)

RestartConnector moves a connector to RESTARTING and records a RESTART_IN_PROGRESS operation that completes after provisionDelay.

func (*InMemoryBackend) Restore 

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

Restore deserializes backend state from a JSON snapshot.

func (*InMemoryBackend) SetAppConfig 

func (b *InMemoryBackend) SetAppConfig(cfg any)

SetAppConfig records the service.AppContext.Config for lazy sibling lookup.

func (*InMemoryBackend) Snapshot 

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

Snapshot serializes backend state to JSON.

func (*InMemoryBackend) TagResource 

func (b *InMemoryBackend) TagResource(resourceArn string, tags map[string]string) error

TagResource adds or replaces tags on a connector, custom plugin, or worker configuration by ARN.

func (*InMemoryBackend) UntagResource 

func (b *InMemoryBackend) UntagResource(resourceArn string, tagKeys []string) error

UntagResource removes tags from a connector, custom plugin, or worker configuration by ARN.

func (*InMemoryBackend) UpdateConnector 

func (b *InMemoryBackend) UpdateConnector(
	connectorArn, currentVersion string,
	update ConnectorUpdate,
) (*Connector, *ConnectorOperation, error)

UpdateConnector updates a connector's capacity or configuration under optimistic lock (currentVersion). Exactly one of update.Capacity or update.ConnectorConfiguration must be set. The connector goes UPDATING and the returned operation UPDATE_IN_PROGRESS until provisionDelay elapses.

type PluginRef 

type PluginRef struct {
	CustomPluginArn string
	Revision        int64
}

PluginRef mirrors types.PluginDescription.CustomPlugin.

type Provider 

type Provider struct{}

Provider implements service.Provider for Amazon MSK Connect.

func (*Provider) Init 

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

Init initializes the MSK Connect service backend and handler.

func (*Provider) Name 

func (p *Provider) Name() string

Name returns the provider name.

type ProvisionedCapacity 

type ProvisionedCapacity struct {
	McuCount    int32
	WorkerCount int32
}

ProvisionedCapacity mirrors types.ProvisionedCapacityDescription.

type S3LogDelivery 

type S3LogDelivery struct {
	Bucket  string
	Prefix  string
	Enabled bool
}

S3LogDelivery mirrors types.S3LogDeliveryDescription.

type Snapshottable 

type Snapshottable interface {
	Snapshot(ctx context.Context) []byte
	Restore(context.Context, []byte) error
}

Snapshottable is an optional interface a StorageBackend may implement to support snapshot/restore for persistence or test isolation.

type StorageBackend 

type StorageBackend interface {
	CreateConnector(accountID, region string, spec ConnectorSpec) (*Connector, error)
	DescribeConnector(connectorArn string) (*Connector, error)
	ListConnectors(namePrefix, nextToken string, maxResults int) ([]*Connector, string, error)
	UpdateConnector(
		connectorArn, currentVersion string,
		update ConnectorUpdate,
	) (*Connector, *ConnectorOperation, error)
	DeleteConnector(connectorArn, currentVersion string) (*Connector, error)
	RestartConnector(connectorArn string, onlyFailedTasks bool) (*Connector, *ConnectorOperation, error)
	DescribeConnectorOperation(operationArn string) (*ConnectorOperation, error)
	ListConnectorOperations(connectorArn, nextToken string, maxResults int) ([]*ConnectorOperation, string, error)

	CreateCustomPlugin(
		accountID, region, name, description, contentType, bucketArn, fileKey, objectVersion string,
		tags map[string]string,
	) (*CustomPlugin, error)
	DescribeCustomPlugin(customPluginArn string) (*CustomPlugin, error)
	ListCustomPlugins(namePrefix, nextToken string, maxResults int) ([]*CustomPlugin, string, error)
	DeleteCustomPlugin(customPluginArn string) (*CustomPlugin, error)

	CreateWorkerConfiguration(
		accountID, region, name, description, propertiesFileContent string,
		tags map[string]string,
	) (*WorkerConfiguration, error)
	DescribeWorkerConfiguration(workerConfigurationArn string) (*WorkerConfiguration, error)
	ListWorkerConfigurations(namePrefix, nextToken string, maxResults int) ([]*WorkerConfiguration, string, error)
	DeleteWorkerConfiguration(workerConfigurationArn string) (*WorkerConfiguration, error)

	TagResource(resourceArn string, tags map[string]string) error
	UntagResource(resourceArn string, tagKeys []string) error
	ListTagsForResource(resourceArn string) (map[string]string, error)

	Reset()
}

StorageBackend is the interface for the MSK Connect backend.

type Vpc 

type Vpc struct {
	SecurityGroups []string
	Subnets        []string
}

Vpc mirrors types.VpcDescription.

type WorkerConfigRef 

type WorkerConfigRef struct {
	Arn      string
	Revision int64
}

WorkerConfigRef mirrors types.WorkerConfigurationDescription.

type WorkerConfigRevision 

type WorkerConfigRevision struct {
	CreationTime          time.Time
	Description           string
	PropertiesFileContent string
	Revision              int64
}

WorkerConfigRevision is the persisted representation of a worker configuration revision. This backend only ever creates revision 1 -- UpdateWorkerConfiguration (which would create new revisions) is not part of the AWS API surface this backend implements.

type WorkerConfiguration 

type WorkerConfiguration struct {
	CreationTime   time.Time
	PendingUntil   time.Time
	Tags           map[string]string
	Name           string
	ARN            string
	Description    string
	State          string
	LatestRevision WorkerConfigRevision
}

WorkerConfiguration is the persisted representation of an MSK Connect worker configuration.

type WorkerLogDelivery 

type WorkerLogDelivery struct {
	CloudWatchLogs *CloudWatchLogsDelivery
	Firehose       *FirehoseDelivery
	S3             *S3LogDelivery
}

WorkerLogDelivery mirrors types.WorkerLogDeliveryDescription.

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.