api

type ACLEntry struct {
	Principal    string // e.g. "User:CN=devuser,..."
	Host         string // e.g. "*" or specific IP
	ResourceType string // "Topic", "Group", "Cluster", etc.
	ResourceName string // resource name or "*"
	PatternType  string // "Literal", "Prefixed", etc.
	Operation    string // "Read", "Write", "Describe", etc.
	Permission   string // "Allow" or "Deny"
}

type ACLFilter struct {
	ResourceType string // "Topic", "Group", ... ("" = any)
	ResourceName string // exact resource name ("" = any)
	PatternType  string // "Literal", "Prefixed", ... ("" = any)
}

type ACLNotFoundError struct {
	Entry ACLEntry
	Cause error
}

type ACLValidationError struct {
	Field  string
	Reason string
	Cause  error
}

type AccessDeniedError struct {
	Resource string
	Name     string // resource name, empty for unnamed/create checks
	Action   string
	Cause    error
}

type AnalysisAlreadyRunningError struct {
	TopicName string
}

type AnalysisProgress struct {
	StartTime        time.Time
	ProcessedOffsets int64
	TotalOffsets     int64
	MessagesScanned  int64
	BytesScanned     int64
}

type AnalysisState string

type AuthenticationError struct {
	Message string
	Method  string
}

type AuthorizationError struct {
	Message  string
	Resource string
	Action   string
}

type BrokerConfigEntry struct {
	Name      string
	Value     string
	Source    string
	Sensitive bool
	ReadOnly  bool
	Synonyms  []BrokerConfigSynonym
}

type BrokerConfigSynonym struct {
	Name   string
	Value  string
	Source string
}

type BrokerDiskUsage struct {
	BrokerID         int32
	TotalSegmentSize int64
	SegmentCount     int
}

type BrokerInfo struct {
	ID           int32
	Host         string
	Port         int32
	Rack         string
	IsController bool
}

type BrokerLogDir struct {
	Path   string
	Error  string
	Topics []BrokerLogDirTopic
}

type BrokerLogDirPartition struct {
	Partition int32
	Size      int64
	OffsetLag int64
}

type BrokerLogDirTopic struct {
	Topic      string
	Partitions []BrokerLogDirPartition
}

type BrokerMetrics struct {
	ID             int32
	LeaderCount    int
	ReplicaCount   int
	SegmentSize    int64
	BytesInPerSec  float64 // RateUnknown unless a metrics endpoint is configured
	BytesOutPerSec float64 // RateUnknown unless a metrics endpoint is configured
}

type BrokerNotFoundError struct {
	BrokerID int32
	Cause    error
}

type BrokerStats struct {
	SegmentSize        int64
	SegmentCount       int
	LeaderCount        int
	ReplicaCount       int
	InSyncReplicaCount int
	ReplicaSkew        *float64
	LeaderSkew         *float64
}

type BrokerSummary struct {
	BrokerCount      int
	ControllerID     *int32
	ClusterVersion   string
	OnlinePartitions int
	TotalPartitions  int
	UnderReplicated  int
	InSyncReplicas   int
	TotalReplicas    int
	OutOfSync        int
	ControllerType   string // "KRaft" | "ZooKeeper" | "Unknown"
}

type BrowseEvent struct {
	Phase       BrowsePhase
	Description string
	Stats       BrowseStats
	Done        bool
}

type BrowsePhase string

type BrowseSession struct {
	Topic      string
	Seek       SeekMode
	Partitions []int32
	Filter     string
	KeySerde   string
	ValueSerde string
	PageSize   int

	// NextPositions holds the next offset to poll per partition. It is updated
	// by RecordPage as pages are consumed.
	NextPositions map[int32]int64
	// contains filtered or unexported fields
}

type BrowseStats struct {
	MessagesConsumed int64
	BytesConsumed    int64
	FilterErrors     int64
	ElapsedMs        int64
}

type Capability string

type CleanupPolicyError struct {
	TopicName string
	Policy    string
}

type ClientQuotaEntity struct {
	User     *string
	ClientID *string
	IP       *string
}

type ClientQuotaEntry struct {
	Entity ClientQuotaEntity
	Quotas map[string]float64
}

type ClusterInfo struct {
	Name              string
	Brokers           []string
	SchemaRegistryURL string
	IsCurrent         bool
	ReadOnly          bool
}

type ClusterMetrics struct {
	Cluster          string
	CollectedAt      time.Time
	BrokerCount      int
	TopicCount       int
	PartitionCount   int
	MessageCount     int64
	MessagesInPerSec float64 // RateUnknown until a second collection cycle establishes a delta
	BytesInPerSec    float64 // RateUnknown unless a metrics endpoint is configured
	BytesOutPerSec   float64 // RateUnknown unless a metrics endpoint is configured
	Topics           []TopicMetrics
	Brokers          []BrokerMetrics
}

type ClusterNotFoundError struct {
	Name      string
	Available []string // known cluster names, when the caller has them handy
	Cause     error
}

type ClusterOverview struct {
	Name                 string
	Status               ClusterStatus
	LastError            string
	BrokerCount          int
	OnlinePartitionCount int
	TopicCount           int
	BytesInPerSec        float64
	BytesOutPerSec       float64
	MessagesInPerSec     float64
	ReadOnly             bool
	Version              string
	Capabilities         []Capability
}

type ClusterReadOnlyError struct {
	Cluster   string
	Operation string
	Cause     error
}

type ClusterStatistics struct {
	BrokerCount               int
	ControllerID              int32
	OnlinePartitions          int
	OfflinePartitions         int
	InSyncReplicas            int
	OutOfSyncReplicas         int
	UnderReplicatedPartitions int
	DiskUsage                 []BrokerDiskUsage
	Version                   string
	CoordinationType          string // "kraft", "zookeeper", or "unknown"
}

type ClusterStatus string

type ClusterValidation struct {
	Cluster string
	Results []ValidationResult
}

type CompatibilityLevel string

type ConnectCluster struct {
	Name           string
	Address        string
	Version        string
	Commit         string
	KafkaClusterID string
	Reachable      bool

	// Aggregated statistics (only set when withStats=true).
	ConnectorCount       int
	FailedConnectorCount int
	TaskCount            int
	FailedTaskCount      int
}

type ConnectClusterNotFoundError struct {
	Connect string // the Connect cluster name that was not found
	Cluster string // the active Kafka cluster/context it was looked up in
	Cause   error
}

type ConnectionError struct {
	Message string
	Cause   error
}

type Connector struct {
	ConnectCluster  string
	Name            string
	Class           string
	Type            ConnectorType
	Topics          []string
	State           string
	WorkerID        string
	Trace           string
	TaskCount       int
	FailedTaskCount int
	ConsumerGroup   string // derived for sink connectors from the configured pattern
}

type ConnectorAlreadyExistsError struct {
	Connector string
	Connect   string
	Cause     error
}

type ConnectorConfigKeyValidation struct {
	Name              string
	Value             string
	Errors            []string
	RecommendedValues []string
	Visible           bool
}

type ConnectorDetails struct {
	ConnectCluster string
	Name           string
	Class          string
	Type           ConnectorType
	Config         map[string]string
	State          string
	WorkerID       string
	Trace          string
	Tasks          []ConnectorTask
	Topics         []string
	ConsumerGroup  string
}

type ConnectorNotFoundError struct {
	Connector string
	Connect   string
	Cause     error
}

type ConnectorNotStoppedError struct {
	Connector string
	Connect   string
	State     string
	Cause     error
}

type ConnectorPlugin struct {
	Class   string
	Type    string // source | sink
	Version string
}

type ConnectorTask struct {
	ID       int
	WorkerID string
	State    string // RUNNING | FAILED | PAUSED | RESTARTING | UNASSIGNED
	Trace    string
}

type ConnectorType string

type ConnectorValidationResult struct {
	Name       string
	ErrorCount int
	Groups     []string
	Configs    []ConnectorConfigKeyValidation
}

type ConsumeFlags struct {
	Follow        bool
	Tail          int32
	OffsetFlag    string
	GroupFlag     string
	LimitMessages int64 // stop after N messages (0 = unlimited, runs until HWM or idle timeout)

	// Typed seek model (MSG-1). When Seek is set it takes precedence over
	// OffsetFlag for per-partition offset resolution; OffsetFlag remains for
	// backward compatibility with existing callers.
	Seek          SeekMode
	SeekOffset    *int64     // required for SeekFromOffset / SeekToOffset
	SeekTimestamp *time.Time // required for SeekFromTimestamp / SeekToTimestamp
	Partitions    []int32    // empty = all partitions
}

type ConsumerGroup struct {
	Name      string
	State     string
	Consumers int

	// Enrichment fields populated lazily by GetConsumerGroupDetails for the
	// visible page of rows. Lag is nil when undefined (no committed offsets).
	MemberCount       int
	TopicCount        int
	Lag               *int64
	CoordinatorID     int32
	PartitionAssignor string
	IsSimple          bool
}

type ConsumerGroupDetail struct {
	GroupID           string
	State             string // canonical GroupState* value
	ProtocolType      string
	PartitionAssignor string
	IsSimple          bool // protocol type is not "consumer"
	CoordinatorID     int32
	Members           []GroupMember
	TopicOffsets      []PartitionOffset
}

type GroupLag struct {
	TotalLag *int64
	PerTopic map[string]*int64
}

type GroupMember struct {
	ConsumerID  string // MemberId assigned by the coordinator
	ClientID    string // ClientId from the member's join request
	Host        string // ClientHost
	Assignments []TopicPartition
}

type GroupNotEmptyError struct {
	GroupID string
	State   string
	Cause   error
}

type GroupNotFoundError struct {
	GroupID string
	Cause   error
}

type InvalidConfigError struct {
	Key    string
	Reason string
	Cause  error
}

type InvalidOffsetResetError struct {
	Reason string
	Cause  error
}

type InvalidReplicationFactorError struct {
	TopicName string
	Reason    string
}

type InvalidSeekError struct {
	Mode   string
	Reason string
	Cause  error
}

type KafkaDataSource interface {
	Init(cfgOption string)
	GetTopics() (map[string]Topic, error)
	// GetTopicNames returns only topic names using a lightweight metadata request.
	// This is significantly faster than GetTopics() on large clusters because it
	// skips per-partition replica details. Use it to show names immediately, then
	// call GetTopics() asynchronously to fill in partition/replication details.
	GetTopicNames() ([]string, error)
	GetContexts() ([]string, error)
	GetContext() string
	SetContext(contextName string) error
	// GetClusterDetails returns configuration details for the named cluster.
	// Used by the context view to display broker addresses and schema registry URL.
	GetClusterDetails(clusterName string) (ClusterInfo, error)
	GetConsumerGroups() ([]ConsumerGroup, error)
	// GetConsumerGroupDetail returns the full description of ONE consumer group,
	// including per-partition committed/end offsets and lag. It describes only
	// the named group (never the whole listing) and returns a GroupNotFoundError
	// when the group id is unknown.
	GetConsumerGroupDetail(groupID string) (ConsumerGroupDetail, error)
	// GetConsumerGroupDetails enriches a batch of groups (the currently visible
	// page of list rows) with real state, member/topic counts, lag, coordinator
	// and assignor. It is a bounded fan-out — pass only the visible names, never
	// the entire cluster. Groups that fail to describe keep state Unknown and nil
	// lag rather than failing the whole batch.
	GetConsumerGroupDetails(groupIDs []string) ([]ConsumerGroup, error)
	// GetConsumerGroupsForTopic returns groups related to a topic: a group is
	// related if any active member is assigned a partition of the topic or the
	// group has committed offsets for it. This inherently fans out and should be
	// invoked only on explicit user request.
	GetConsumerGroupsForTopic(topic string) ([]ConsumerGroup, error)
	// DeleteConsumerGroup deletes a consumer group. It returns a GroupNotEmptyError
	// when the group still has active members and a GroupNotFoundError when unknown.
	DeleteConsumerGroup(groupID string) error
	// DeleteConsumerGroupOffsets deletes the committed offsets of a single topic
	// for a group, leaving other topics' offsets and the group itself intact.
	DeleteConsumerGroupOffsets(groupID string, topic string) error
	// ResetConsumerGroupOffsets resets a group's committed offsets for a topic.
	// The group must be inactive (Empty or Dead); otherwise a GroupNotEmptyError
	// is returned. Invalid requests yield an InvalidOffsetResetError.
	ResetConsumerGroupOffsets(ctx context.Context, req OffsetResetRequest) error
	ConsumeTopic(ctx context.Context, topicName string, flags ConsumeFlags, handleMessage MessageHandlerFunc, onError func(err any)) error
	// ProduceMessage produces a single record to the topic. It validates that
	// the topic exists and that any explicit partition is in range, returning a
	// TopicNotFoundError, PartitionError, or ProduceError respectively.
	ProduceMessage(ctx context.Context, topic string, rec ProduceRecord) error
	// GetTopicMessageCounts fetches approximate message counts for a set of topics.
	// It accepts a map of topicName → numPartitions (already known from GetTopics).
	// Counts are computed as sum(newestOffset - oldestOffset) across all partitions.
	// Returns a best-effort map: topics that fail are omitted rather than failing the whole call.
	GetTopicMessageCounts(topics map[string]int32) (map[string]int64, error)
	// GetSchemas returns all registered schema subjects from the schema registry.
	// It fetches only subject names — no version/ID/type details — so it completes
	// with a single HTTP request even for large registries.
	// Returns an empty slice (not an error) when no schema registry is configured.
	GetSchemas() ([]Schema, error)
	// GetSchemaDetails fetches the latest version metadata (version, id, schemaType)
	// for a specific set of subjects. Used to lazily populate table rows for the
	// current page rather than loading details for all subjects at once.
	GetSchemaDetails(subjects []string) ([]Schema, error)
	// GetSchemaContent fetches the full schema definition string (Avro JSON, Protobuf IDL, etc.)
	// for the given subject. Pass version=0 to retrieve the latest version.
	GetSchemaContent(subject string, version int) (string, error)
	// GetSchemaVersions lists all registered versions of a subject (metadata only;
	// Schema text is left empty — fetch it lazily via GetSchemaContent). Returns a
	// SubjectNotFoundError for an unknown subject and a
	// SchemaRegistryNotConfiguredError when no registry is configured. (SR-4)
	GetSchemaVersions(subject string) ([]SchemaVersion, error)
	// GetGlobalCompatibility returns the registry's global compatibility level. (SR-5)
	GetGlobalCompatibility() (CompatibilityLevel, error)
	// GetSubjectCompatibility returns a subject's effective compatibility level.
	// When the subject has no own setting it falls back to the global level and
	// returns isSubjectSpecific=false. (SR-5)
	GetSubjectCompatibility(subject string) (level CompatibilityLevel, isSubjectSpecific bool, err error)
	// RegisterSchema registers a new schema under the subject, creating the subject
	// if new or a new version otherwise. schemaType may be empty for AVRO. It maps
	// a 409 to SchemaIncompatibleError and a 422 to SchemaValidationError. (SR-7)
	RegisterSchema(subject, schemaText, schemaType string) (Schema, error)
	// CheckSchemaCompatibility tests a candidate schema against the subject's latest
	// version without registering it, returning the verbose messages on failure. (SR-8)
	CheckSchemaCompatibility(subject, schemaText, schemaType string) (compatible bool, messages []string, err error)
	// DeleteSubject deletes all versions of a subject. permanent=true performs a
	// hard delete (requires a prior soft delete). It returns the deleted version
	// numbers. (SR-9)
	DeleteSubject(subject string, permanent bool) ([]int, error)
	// DeleteSchemaVersion deletes a single version. Pass version=-1 to delete the
	// latest version. permanent=true performs a hard delete. (SR-9)
	DeleteSchemaVersion(subject string, version int, permanent bool) error
	// SetGlobalCompatibility sets the registry's global compatibility level. It
	// validates the level before making any HTTP call. (SR-10)
	SetGlobalCompatibility(level CompatibilityLevel) error
	// SetSubjectCompatibility sets a subject's compatibility level. It validates
	// the level before making any HTTP call. (SR-10)
	SetSubjectCompatibility(subject string, level CompatibilityLevel) error
	// GetACLs returns all ACL bindings for the current cluster.
	// Returns an empty slice (not an error) when the broker returns no ACLs
	// or the connected user lacks the DESCRIBE ACL on cluster resources.
	GetACLs() ([]ACLEntry, error)
	// GetACLsFiltered returns ACL bindings matching the given filter. Empty
	// filter fields match any value; GetACLs is the match-any case.
	GetACLsFiltered(filter ACLFilter) ([]ACLEntry, error)
	// CreateACL creates a single ACL binding. The entry is validated (principal
	// format, non-empty resource/operation/permission); an empty PatternType
	// defaults to Literal. Invalid entries yield an ACLValidationError.
	CreateACL(entry ACLEntry) error
	// DeleteACL deletes the ACL binding exactly matching the full entry
	// definition. It returns an ACLNotFoundError when no binding matches.
	DeleteACL(entry ACLEntry) error
	// GetClientQuotas returns all configured client quotas, deterministically
	// ordered (user -> client-id -> ip, absent identifiers last).
	GetClientQuotas() ([]ClientQuotaEntry, error)
	// AlterClientQuotas upserts the quota values for an entity with replace
	// semantics: the submitted map becomes the entity's complete property set
	// (absent properties are cleared); an empty/nil map deletes the entity.
	// A QuotaValidationError is returned when no entity identifier is set.
	AlterClientQuotas(entity ClientQuotaEntity, quotas map[string]float64) error
	// GetMessageSchemaInfo retrieves schema information for a message's key and value
	// Returns nil for non-Avro messages or when schema information is not available
	GetMessageSchemaInfo(keySchemaID, valueSchemaID string) (*MessageSchemaInfo, error)
	// DecodeMessage decodes the raw bytes of a message (e.g. Avro) into human-readable
	// Key/Value strings. If the message is already decoded or has no raw bytes to decode,
	// it is returned unchanged. This is used for lazy, on-demand decoding of visible messages.
	DecodeMessage(ctx context.Context, msg Message) (Message, error)
	// ListSerdes returns the names of the serdes available for decoding message
	// keys/values (for the topic-page serde selector). The UI offers "auto"
	// (auto-detection) in addition to these explicit names.
	ListSerdes() []string
	// GetClusterStatistics collects a fresh detailed snapshot of the named cluster.
	// No caching happens here; the background collector (pkg/cluster) owns caching.
	GetClusterStatistics(ctx context.Context, clusterName string) (ClusterStatistics, error)
	// GetClusterCapabilities probes which optional features the named cluster supports.
	GetClusterCapabilities(ctx context.Context, clusterName string) ([]Capability, error)
	// ValidateClusterConnection independently probes each component (broker, schema
	// registry, ...) of the named cluster and returns one result per component.
	// Works for non-active clusters without switching the active context.
	ValidateClusterConnection(ctx context.Context, clusterName string) ([]ValidationResult, error)

	// GetBrokers returns the brokers currently online in the active cluster,
	// with IsController set on the active controller.
	GetBrokers() ([]BrokerInfo, error)
	// GetBrokerStats returns per-broker partition-distribution / disk statistics
	// plus a cluster-wide BrokerSummary. Skew fields are absent (nil) when the
	// cluster has too few partitions to be meaningful.
	GetBrokerStats() (map[int32]BrokerStats, BrokerSummary, error)
	// GetBrokerLogDirs returns log directories for the given broker IDs. An empty
	// brokerIDs slice means "all brokers"; unknown IDs are silently dropped. On
	// timeout it returns an empty result rather than an error.
	GetBrokerLogDirs(brokerIDs []int32) (map[int32][]BrokerLogDir, error)
	// GetBrokerConfig returns the configuration entries for a broker. Unknown
	// broker IDs yield a BrokerNotFoundError.
	GetBrokerConfig(brokerID int32) ([]BrokerConfigEntry, error)
	// AlterBrokerConfig sets a single dynamic broker config key, preserving other
	// dynamic configs. A cluster rejection yields an InvalidConfigError.
	AlterBrokerConfig(brokerID int32, key, value string) error
	// AlterReplicaLogDir moves a topic-partition replica to a different log dir
	// on the target broker.
	AlterReplicaLogDir(brokerID int32, topic string, partition int32, logDir string) error
	// GetBrokerMetrics returns a JSON metrics snapshot for a broker, or a
	// MetricsNotAvailableError when metrics collection is not available.
	GetBrokerMetrics(brokerID int32) (string, error)

	// GetTopicConfig returns the topic's configuration entries with metadata
	// (default value derived from synonyms, source, sensitive, read-only). It
	// returns an empty slice (not an error) on authorization failure.
	GetTopicConfig(topicName string) ([]TopicConfigEntry, error)
	// GetTopicDetails returns per-partition detail plus topic-wide health
	// metrics. It returns a TopicNotFoundError when the topic does not exist.
	GetTopicDetails(topicName string) (TopicDetails, error)
	// GetTopicSizes returns the on-disk size (leader replicas only) per topic.
	// Best-effort: topics that fail are omitted from the result map.
	GetTopicSizes(topicNames []string) (map[string]int64, error)
	// CreateTopic creates a topic. A replicationFactor of -1 requests the
	// cluster default. Empty-valued config entries must be stripped by the
	// caller. It polls metadata until the topic is visible.
	CreateTopic(name string, numPartitions int32, replicationFactor int16, configs map[string]*string) error
	// DeleteTopic deletes a topic.
	DeleteTopic(name string) error
	// IsTopicDeletionEnabled reports whether the cluster permits topic deletion
	// (delete.topic.enable). Missing/unparseable config defaults to true.
	IsTopicDeletionEnabled() (bool, error)
	// UpdateTopicConfig incrementally alters the given topic config entries,
	// leaving unrelated entries untouched.
	UpdateTopicConfig(name string, entries map[string]*string) error
	// IncreasePartitions raises the topic's total partition count. It rejects a
	// decrease (PartitionDecreaseError) or a no-op (PartitionNoopError).
	IncreasePartitions(name string, totalCount int32) error
	// PurgeTopicMessages deletes records up to the high-watermark. A partition of
	// -1 purges all partitions. It returns a CleanupPolicyError for compact-only
	// topics.
	PurgeTopicMessages(name string, partition int32) error
	// RecreateTopic deletes and re-creates a topic preserving its partition
	// count, replication factor, and non-default configs.
	RecreateTopic(name string) error
	// ChangeReplicationFactor changes the replication factor of every partition
	// via a computed balanced reassignment across online brokers.
	ChangeReplicationFactor(name string, newFactor int16) error

	// StartTopicAnalysis begins a background scan+aggregation of the topic. It
	// returns an AnalysisAlreadyRunningError if one is already in progress and a
	// TopicNotFoundError for an unknown topic.
	StartTopicAnalysis(ctx context.Context, topicName string) error
	// GetTopicAnalysis returns the latest analysis (running/completed/failed) for
	// the topic, or (nil, nil) when none has ever been started.
	GetTopicAnalysis(topicName string) (*TopicAnalysis, error)
	// CancelTopicAnalysis cancels a running analysis, releasing its resources and
	// retaining no result.
	CancelTopicAnalysis(topicName string) error

	// GetConnectClusters lists the Connect clusters configured for the active
	// Kafka cluster. Each cluster is probed for its runtime info (version, commit,
	// kafka_cluster_id); an unreachable cluster is still listed with Reachable=false
	// and empty runtime fields (never failing the whole call). When withStats is
	// true it additionally computes per-cluster connector/task counts, which is
	// more expensive as it fetches connector status.
	GetConnectClusters(withStats bool) ([]ConnectCluster, error)
	// GetConnectorNames returns the connector names of a single Connect cluster
	// (fast, names only). Unknown cluster name yields a ConnectClusterNotFoundError.
	GetConnectorNames(connect string) ([]string, error)
	// GetConnectors returns connectors aggregated across all configured Connect
	// clusters, each enriched with type, topics, status and task counts. Unreachable
	// clusters are omitted (their connectors dropped) rather than failing the call.
	GetConnectors() ([]Connector, error)
	// GetConnectorDetails combines a connector's config, status, tasks and topics.
	// Config values are masked. Missing status yields state UNASSIGNED with an
	// empty task list rather than an error.
	GetConnectorDetails(connect, name string) (ConnectorDetails, error)
	// CreateConnector creates a connector from the given config. A duplicate name
	// yields a ConnectorAlreadyExistsError.
	CreateConnector(connect, name string, config map[string]string) (Connector, error)
	// UpdateConnectorConfig replaces a connector's configuration. The supplied
	// config is sent unmasked; masked placeholders must be resolved by the caller.
	UpdateConnectorConfig(connect, name string, config map[string]string) (Connector, error)
	// DeleteConnector deletes a connector. Unknown connector yields a
	// ConnectorNotFoundError.
	DeleteConnector(connect, name string) error
	// PauseConnector pauses a connector and its tasks.
	PauseConnector(connect, name string) error
	// ResumeConnector resumes a paused/stopped connector.
	ResumeConnector(connect, name string) error
	// StopConnector stops a connector (STOPPED state), a prerequisite for offset
	// reset.
	StopConnector(connect, name string) error
	// RestartConnector restarts the connector instance (not its tasks).
	RestartConnector(connect, name string) error
	// RestartConnectorTask restarts a single task of a connector by task id.
	RestartConnectorTask(connect, name string, taskID int) error
	// ResetConnectorOffsets resets a connector's offsets. The connector must be in
	// the STOPPED state; otherwise a ConnectorNotStoppedError is returned without
	// calling the API. Unknown connector yields a ConnectorNotFoundError.
	ResetConnectorOffsets(connect, name string) error
	// GetConnectorPlugins lists the connector plugins installed on a Connect cluster.
	GetConnectorPlugins(connect string) ([]ConnectorPlugin, error)
	// ValidateConnectorConfig validates a candidate configuration against a named
	// plugin class and returns the per-field validation outcome.
	ValidateConnectorConfig(connect, pluginClass string, config map[string]string) (ConnectorValidationResult, error)

	// ListKsqlStreams lists the ksqlDB streams for the active cluster (posts LIST
	// STREAMS; to /ksql). Not-configured yields a KsqlNotConfiguredError.
	ListKsqlStreams() ([]KsqlStream, error)
	// ListKsqlTables lists the ksqlDB tables for the active cluster (posts LIST
	// TABLES; to /ksql). Not-configured yields a KsqlNotConfiguredError.
	ListKsqlTables() ([]KsqlTable, error)
	// ExecuteKsql validates and executes a single ksqlDB statement, delivering all
	// outcomes (schema, data rows, statement responses, and errors) as
	// KsqlResultTables on the returned channel. Validation runs first and routing
	// (SELECT -> streaming query; everything else -> statement) is internal.
	// SELECT queries stream a schema table followed by one table per data row until
	// the query ends or ctx is cancelled; cancelling ctx terminates the server-side
	// query and closes the channel. A KsqlNotConfiguredError is returned (with a
	// nil channel) when no endpoint is configured.
	ExecuteKsql(ctx context.Context, sql string, props map[string]string) (<-chan KsqlResultTable, error)
}

type KsqlNoInstancesError struct {
	Configured int
	Cause      error
}

type KsqlNotConfiguredError struct {
	Cause error
}

type KsqlResultTable struct {
	Title   string
	Columns []string
	Rows    [][]string
	IsError bool
}

type KsqlServerError struct {
	StatusCode int
	ErrorCode  int
	Message    string
	Raw        string
	Cause      error
}

type KsqlStream struct {
	Name        string
	Topic       string
	KeyFormat   string
	ValueFormat string
}

type KsqlTable struct {
	Name        string
	Topic       string
	KeyFormat   string
	ValueFormat string
	Windowed    bool
}

type LogDirNotFoundError struct {
	Path  string
	Cause error
}

type Message struct {
	Key   string
	Value string
	// RawKey and RawValue hold the original Kafka bytes for Avro-encoded messages.
	// They are populated at consumption time and decoded lazily via DecodeMessage.
	RawKey        []byte
	RawValue      []byte
	Offset        int64
	Partition     int32
	KeySchemaID   string
	ValueSchemaID string
	Headers       []MessageHeader
	// Timestamp is the message's broker timestamp. Zero when unknown. Used by the
	// analysis engine for time-range and hourly-bucket aggregation.
	Timestamp time.Time
	// TimestampType records how Timestamp was assigned (create vs log-append).
	TimestampType TimestampType
	// KeySize/ValueSize hold the raw byte sizes of the key/value. They are nil
	// when the key/value is null (as opposed to an empty byte slice, size 0).
	KeySize   *int
	ValueSize *int
	// HeadersSize is the summed byte size of all header keys and values.
	HeadersSize int
	// KeyNull/ValueNull distinguish a null key/value from an empty one.
	KeyNull   bool
	ValueNull bool
	// KeySerde/ValueSerde name the serde used to render the key/value. Until the
	// serde framework lands these carry the active decoder name (e.g. "avro").
	KeySerde   string
	ValueSerde string
}

type MessageHandlerFunc func(msg Message)

type MessageHeader struct {
	Key   string
	Value string
}

type MessageHeaders []MessageHeader

type MessageSchemaInfo struct {
	KeySchema   *SchemaInfo `json:"keySchema,omitempty"`
	ValueSchema *SchemaInfo `json:"valueSchema,omitempty"`
}

type MetadataTimeoutError struct {
	TopicName string
	Cause     error
}

type MetricPoint struct {
	Time  time.Time
	Value float64
}

type MetricsNotAvailableError struct {
	BrokerID int32
	Cause    error
}

type MetricsNotConfiguredError struct {
	Cluster string
	Cause   error
}

type NotSupportedError struct {
	Operation string
}

type OffsetResetMode string

type OffsetResetRequest struct {
	GroupID          string
	Topic            string
	Mode             OffsetResetMode
	Partitions       []int32
	Timestamp        *time.Time      // required for OffsetResetTimestamp
	PartitionOffsets map[int32]int64 // required for OffsetResetExplicit
}

type PartitionAnalysis struct {
	Partition    int32
	MessageCount int64
	MinOffset    int64
	MaxOffset    int64
}

type PartitionDecreaseError struct {
	TopicName string
	Current   int32
	Requested int32
}

type PartitionError struct {
	Message     string
	TopicName   string
	PartitionID int32
}

type PartitionInfo struct {
	ID              int32
	Leader          int32
	Replicas        []int32
	ISR             []int32
	OfflineReplicas []int32
	EarliestOffset  int64
	LatestOffset    int64
}

type PartitionNoopError struct {
	TopicName string
	Current   int32
}

type PartitionOffset struct {
	Topic           string
	Partition       int32
	CommittedOffset *int64
	EndOffset       int64
	Lag             *int64
	MemberID        string // consumer assigned to this partition, if any
	MemberHost      string
}

type ProduceError struct {
	Topic  string
	Reason string
	Cause  error
}

type ProduceRecord struct {
	Key       []byte
	Value     []byte
	Headers   []MessageHeader
	Partition *int32 // nil = let the partitioner choose
}

type QuotaValidationError struct {
	Reason string
	Cause  error
}

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

type RecreateTimeoutError struct {
	TopicName string
	Cause     error
}

type Schema struct {
	Subject    string `json:"subject"`
	Version    int    `json:"version"`
	ID         int    `json:"id"`
	SchemaType string `json:"schemaType"` // AVRO, PROTOBUF, JSON — empty means AVRO
	// Compatibility is the subject's effective compatibility level (falling back
	// to the global level when the subject has no own setting). Empty when not
	// resolved. (SR-6)
	Compatibility string `json:"compatibility,omitempty"`
}

type SchemaIncompatibleError struct {
	Subject string
	Message string
	Cause   error
}

type SchemaInfo struct {
	ID         int    `json:"id"`
	Schema     string `json:"schema"`
	Subject    string `json:"subject"`
	Version    int    `json:"version"`
	RecordName string `json:"recordName"` // The type name (e.g., AddedItemToChartEvent)
}

type SchemaRegistryNotConfiguredError struct {
	Cause error
}

type SchemaValidationError struct {
	Message string
	Cause   error
}

type SchemaVersion struct {
	Version    int    `json:"version"`
	ID         int    `json:"id"`
	SchemaType string `json:"schemaType"` // AVRO, PROTOBUF, JSON — empty means AVRO
	Schema     string `json:"schema"`
}

type SchemaVersionNotFoundError struct {
	Subject string
	Version int
	Cause   error
}

type SeekMode string

type SizeDistribution struct {
	Count int64
	Sum   int64
	Min   int64
	Max   int64
	Avg   float64
	P50   int64
	P75   int64
	P95   int64
	P99   int64
	P999  int64
}

type SubjectNotFoundError struct {
	Subject string
	Cause   error
}

type Summary struct {
	Min   float64
	Max   float64
	Avg   float64
	Last  float64
	Count int
	OK    bool
}

type TimeSeries struct {
	Points []MetricPoint
}

type TimeoutError struct {
	Message string
	Timeout string
}

type TimestampType string

type Topic struct {
	// NumPartitions contains the number of partitions to create in the topic
	NumPartitions int32
	// ReplicationFactor contains the number of replicas to create for each partition
	ReplicationFactor int16
	// ReplicaAssignment contains the manual partition assignment, or the empty
	// array if we are using automatic assignment.
	ReplicaAssignment map[int32][]int32
	// ConfigEntries contains the custom topic configurations to set.
	ConfigEntries map[string]*string
	// Num of messages in the topic across all partitions
	MessageCount int64
}

type TopicAlreadyExistsError struct {
	TopicName string
	Cause     error
}

type TopicAnalysis struct {
	Topic    string
	State    AnalysisState
	Progress AnalysisProgress
	Result   *TopicAnalysisResult
	Err      string
	ErrAt    time.Time
}

type TopicAnalysisResult struct {
	Topic                string
	MessageCount         int64
	MinOffset            int64
	MaxOffset            int64
	MinTimestamp         time.Time
	MaxTimestamp         time.Time
	NullKeys             int64
	NullValues           int64
	ApproxDistinctKeys   int64
	ApproxDistinctValues int64
	KeySize              SizeDistribution
	ValueSize            SizeDistribution
	// HourlyCounts maps a Unix-hour bucket (seconds truncated to the hour) to a
	// message count, retained only for the last 14 days.
	HourlyCounts map[int64]int64
	Partitions   []PartitionAnalysis
	CompletedAt  time.Time
}

type TopicConfigEntry struct {
	Name      string
	Value     string
	Default   string
	Source    string
	Sensitive bool
	ReadOnly  bool
}

type TopicDeletionDisabledError struct {
	TopicName string
}

type TopicDetails struct {
	Name                      string
	Partitions                []PartitionInfo
	ReplicationFactor         int16
	IsInternal                bool
	UnderReplicatedPartitions int
	InSyncReplicas            int
	TotalReplicas             int
}

type TopicError struct {
	Message   string
	TopicName string
}

type TopicMetrics struct {
	Name             string
	PartitionCount   int32
	MessageCount     int64
	MessagesInPerSec float64 // derived from message-count deltas; RateUnknown until a prior sample exists
}

type TopicNotFoundError struct {
	TopicName string
	Cause     error
}

type TopicPartition struct {
	Topic     string
	Partition int32
}

type TopicValidationError struct {
	TopicName string
	Reason    string
	Cause     error
}

type ValidationReport struct {
	Clusters []ClusterValidation
}

type ValidationResult struct {
	Component string // "broker", "schema-registry", "tls", "connect:<name>", "ksql", "metrics"
	OK        bool
	Err       string
}