- + 00version.go - + adminapi.go - + adminoptions.go - + build_darwin.go - + config.go - + consumer.go - + context.go - + error.go - + error_gen.go - + event.go - + generated_errors.go - + handle.go - + header.go - + kafka.go - + log.go - + message.go - + metadata.go - + misc.go - + mockcluster.go - + offset.go - + producer.go - + testhelpers.go - + time.go @@ -1435,87 +1435,87 @@
const (
// ResourceUnknown - Unknown
- ResourceUnknown = ResourceType(C.RD_KAFKA_RESOURCE_UNKNOWN)
+ ResourceUnknown = ResourceType(C.RD_KAFKA_RESOURCE_UNKNOWN)
// ResourceAny - match any resource type (DescribeConfigs)
- ResourceAny = ResourceType(C.RD_KAFKA_RESOURCE_ANY)
+ ResourceAny = ResourceType(C.RD_KAFKA_RESOURCE_ANY)
// ResourceTopic - Topic
- ResourceTopic = ResourceType(C.RD_KAFKA_RESOURCE_TOPIC)
+ ResourceTopic = ResourceType(C.RD_KAFKA_RESOURCE_TOPIC)
// ResourceGroup - Group
- ResourceGroup = ResourceType(C.RD_KAFKA_RESOURCE_GROUP)
+ ResourceGroup = ResourceType(C.RD_KAFKA_RESOURCE_GROUP)
// ResourceBroker - Broker
- ResourceBroker = ResourceType(C.RD_KAFKA_RESOURCE_BROKER)
+ ResourceBroker = ResourceType(C.RD_KAFKA_RESOURCE_BROKER)
)
const (
// ConfigSourceUnknown is the default value
- ConfigSourceUnknown = ConfigSource(C.RD_KAFKA_CONFIG_SOURCE_UNKNOWN_CONFIG)
+ ConfigSourceUnknown = ConfigSource(C.RD_KAFKA_CONFIG_SOURCE_UNKNOWN_CONFIG)
// ConfigSourceDynamicTopic is dynamic topic config that is configured for a specific topic
- ConfigSourceDynamicTopic = ConfigSource(C.RD_KAFKA_CONFIG_SOURCE_DYNAMIC_TOPIC_CONFIG)
+ ConfigSourceDynamicTopic = ConfigSource(C.RD_KAFKA_CONFIG_SOURCE_DYNAMIC_TOPIC_CONFIG)
// ConfigSourceDynamicBroker is dynamic broker config that is configured for a specific broker
- ConfigSourceDynamicBroker = ConfigSource(C.RD_KAFKA_CONFIG_SOURCE_DYNAMIC_BROKER_CONFIG)
+ ConfigSourceDynamicBroker = ConfigSource(C.RD_KAFKA_CONFIG_SOURCE_DYNAMIC_BROKER_CONFIG)
// ConfigSourceDynamicDefaultBroker is dynamic broker config that is configured as default for all brokers in the cluster
- ConfigSourceDynamicDefaultBroker = ConfigSource(C.RD_KAFKA_CONFIG_SOURCE_DYNAMIC_DEFAULT_BROKER_CONFIG)
+ ConfigSourceDynamicDefaultBroker = ConfigSource(C.RD_KAFKA_CONFIG_SOURCE_DYNAMIC_DEFAULT_BROKER_CONFIG)
// ConfigSourceStaticBroker is static broker config provided as broker properties at startup (e.g. from server.properties file)
- ConfigSourceStaticBroker = ConfigSource(C.RD_KAFKA_CONFIG_SOURCE_STATIC_BROKER_CONFIG)
+ ConfigSourceStaticBroker = ConfigSource(C.RD_KAFKA_CONFIG_SOURCE_STATIC_BROKER_CONFIG)
// ConfigSourceDefault is built-in default configuration for configs that have a default value
- ConfigSourceDefault = ConfigSource(C.RD_KAFKA_CONFIG_SOURCE_DEFAULT_CONFIG)
+ ConfigSourceDefault = ConfigSource(C.RD_KAFKA_CONFIG_SOURCE_DEFAULT_CONFIG)
)
const (
// ResourcePatternTypeUnknown is a resource pattern type not known or not set.
- ResourcePatternTypeUnknown = ResourcePatternType(C.RD_KAFKA_RESOURCE_PATTERN_UNKNOWN)
+ ResourcePatternTypeUnknown = ResourcePatternType(C.RD_KAFKA_RESOURCE_PATTERN_UNKNOWN)
// ResourcePatternTypeAny matches any resource, used for lookups.
- ResourcePatternTypeAny = ResourcePatternType(C.RD_KAFKA_RESOURCE_PATTERN_ANY)
+ ResourcePatternTypeAny = ResourcePatternType(C.RD_KAFKA_RESOURCE_PATTERN_ANY)
// ResourcePatternTypeMatch will perform pattern matching
- ResourcePatternTypeMatch = ResourcePatternType(C.RD_KAFKA_RESOURCE_PATTERN_MATCH)
+ ResourcePatternTypeMatch = ResourcePatternType(C.RD_KAFKA_RESOURCE_PATTERN_MATCH)
// ResourcePatternTypeLiteral matches a literal resource name
- ResourcePatternTypeLiteral = ResourcePatternType(C.RD_KAFKA_RESOURCE_PATTERN_LITERAL)
+ ResourcePatternTypeLiteral = ResourcePatternType(C.RD_KAFKA_RESOURCE_PATTERN_LITERAL)
// ResourcePatternTypePrefixed matches a prefixed resource name
- ResourcePatternTypePrefixed = ResourcePatternType(C.RD_KAFKA_RESOURCE_PATTERN_PREFIXED)
+ ResourcePatternTypePrefixed = ResourcePatternType(C.RD_KAFKA_RESOURCE_PATTERN_PREFIXED)
)
const (
// ACLOperationUnknown represents an unknown or unset operation
- ACLOperationUnknown = ACLOperation(C.RD_KAFKA_ACL_OPERATION_UNKNOWN)
+ ACLOperationUnknown = ACLOperation(C.RD_KAFKA_ACL_OPERATION_UNKNOWN)
// ACLOperationAny in a filter, matches any ACLOperation
- ACLOperationAny = ACLOperation(C.RD_KAFKA_ACL_OPERATION_ANY)
+ ACLOperationAny = ACLOperation(C.RD_KAFKA_ACL_OPERATION_ANY)
// ACLOperationAll represents all the operations
- ACLOperationAll = ACLOperation(C.RD_KAFKA_ACL_OPERATION_ALL)
+ ACLOperationAll = ACLOperation(C.RD_KAFKA_ACL_OPERATION_ALL)
// ACLOperationRead a read operation
- ACLOperationRead = ACLOperation(C.RD_KAFKA_ACL_OPERATION_READ)
+ ACLOperationRead = ACLOperation(C.RD_KAFKA_ACL_OPERATION_READ)
// ACLOperationWrite represents a write operation
- ACLOperationWrite = ACLOperation(C.RD_KAFKA_ACL_OPERATION_WRITE)
+ ACLOperationWrite = ACLOperation(C.RD_KAFKA_ACL_OPERATION_WRITE)
// ACLOperationCreate represents a create operation
- ACLOperationCreate = ACLOperation(C.RD_KAFKA_ACL_OPERATION_CREATE)
+ ACLOperationCreate = ACLOperation(C.RD_KAFKA_ACL_OPERATION_CREATE)
// ACLOperationDelete represents a delete operation
- ACLOperationDelete = ACLOperation(C.RD_KAFKA_ACL_OPERATION_DELETE)
+ ACLOperationDelete = ACLOperation(C.RD_KAFKA_ACL_OPERATION_DELETE)
// ACLOperationAlter represents an alter operation
- ACLOperationAlter = ACLOperation(C.RD_KAFKA_ACL_OPERATION_ALTER)
+ ACLOperationAlter = ACLOperation(C.RD_KAFKA_ACL_OPERATION_ALTER)
// ACLOperationDescribe represents a describe operation
- ACLOperationDescribe = ACLOperation(C.RD_KAFKA_ACL_OPERATION_DESCRIBE)
+ ACLOperationDescribe = ACLOperation(C.RD_KAFKA_ACL_OPERATION_DESCRIBE)
// ACLOperationClusterAction represents a cluster action operation
- ACLOperationClusterAction = ACLOperation(C.RD_KAFKA_ACL_OPERATION_CLUSTER_ACTION)
+ ACLOperationClusterAction = ACLOperation(C.RD_KAFKA_ACL_OPERATION_CLUSTER_ACTION)
// ACLOperationDescribeConfigs represents a describe configs operation
- ACLOperationDescribeConfigs = ACLOperation(C.RD_KAFKA_ACL_OPERATION_DESCRIBE_CONFIGS)
+ ACLOperationDescribeConfigs = ACLOperation(C.RD_KAFKA_ACL_OPERATION_DESCRIBE_CONFIGS)
// ACLOperationAlterConfigs represents an alter configs operation
- ACLOperationAlterConfigs = ACLOperation(C.RD_KAFKA_ACL_OPERATION_ALTER_CONFIGS)
+ ACLOperationAlterConfigs = ACLOperation(C.RD_KAFKA_ACL_OPERATION_ALTER_CONFIGS)
// ACLOperationIdempotentWrite represents an idempotent write operation
- ACLOperationIdempotentWrite = ACLOperation(C.RD_KAFKA_ACL_OPERATION_IDEMPOTENT_WRITE)
+ ACLOperationIdempotentWrite = ACLOperation(C.RD_KAFKA_ACL_OPERATION_IDEMPOTENT_WRITE)
)
const (
// ACLPermissionTypeUnknown represents an unknown ACLPermissionType
- ACLPermissionTypeUnknown = ACLPermissionType(C.RD_KAFKA_ACL_PERMISSION_TYPE_UNKNOWN)
+ ACLPermissionTypeUnknown = ACLPermissionType(C.RD_KAFKA_ACL_PERMISSION_TYPE_UNKNOWN)
// ACLPermissionTypeAny in a filter, matches any ACLPermissionType
- ACLPermissionTypeAny = ACLPermissionType(C.RD_KAFKA_ACL_PERMISSION_TYPE_ANY)
+ ACLPermissionTypeAny = ACLPermissionType(C.RD_KAFKA_ACL_PERMISSION_TYPE_ANY)
// ACLPermissionTypeDeny disallows access
- ACLPermissionTypeDeny = ACLPermissionType(C.RD_KAFKA_ACL_PERMISSION_TYPE_DENY)
+ ACLPermissionTypeDeny = ACLPermissionType(C.RD_KAFKA_ACL_PERMISSION_TYPE_DENY)
// ACLPermissionTypeAllow grants access
- ACLPermissionTypeAllow = ACLPermissionType(C.RD_KAFKA_ACL_PERMISSION_TYPE_ALLOW)
+ ACLPermissionTypeAllow = ACLPermissionType(C.RD_KAFKA_ACL_PERMISSION_TYPE_ALLOW)
)
const (
// TimestampNotAvailable indicates no timestamp was set, or not available due to lacking broker support
- TimestampNotAvailable = TimestampType(C.RD_KAFKA_TIMESTAMP_NOT_AVAILABLE)
+ TimestampNotAvailable = TimestampType(C.RD_KAFKA_TIMESTAMP_NOT_AVAILABLE)
// TimestampCreateTime indicates timestamp set by producer (source time)
- TimestampCreateTime = TimestampType(C.RD_KAFKA_TIMESTAMP_CREATE_TIME)
+ TimestampCreateTime = TimestampType(C.RD_KAFKA_TIMESTAMP_CREATE_TIME)
// TimestampLogAppendTime indicates timestamp set set by broker (store time)
- TimestampLogAppendTime = TimestampType(C.RD_KAFKA_TIMESTAMP_LOG_APPEND_TIME)
+ TimestampLogAppendTime = TimestampType(C.RD_KAFKA_TIMESTAMP_LOG_APPEND_TIME)
)
const (
// PurgeInFlight purges messages in-flight to or from the broker.
@@ -1523,17 +1523,17 @@
// broker, making it impossible for the application to know if these
// messages were successfully delivered or not.
// Retrying these messages may lead to duplicates.
- PurgeInFlight = int(C.RD_KAFKA_PURGE_F_INFLIGHT)
+ PurgeInFlight = int(C.RD_KAFKA_PURGE_F_INFLIGHT)
// PurgeQueue Purge messages in internal queues.
- PurgeQueue = int(C.RD_KAFKA_PURGE_F_QUEUE)
+ PurgeQueue = int(C.RD_KAFKA_PURGE_F_QUEUE)
// PurgeNonBlocking Don't wait for background thread queue purging to finish.
- PurgeNonBlocking = int(C.RD_KAFKA_PURGE_F_NON_BLOCKING)
+ PurgeNonBlocking = int(C.RD_KAFKA_PURGE_F_NON_BLOCKING)
)
const (
// AlterOperationSet sets/overwrites the configuration setting.
- AlterOperationSet = iota
+ AlterOperationSet = iota
)
LibrdkafkaLinkInfo explains how librdkafka was linked to the Go client @@ -1542,48 +1542,48 @@
OffsetBeginning represents the earliest offset (logical)
-const OffsetBeginning = Offset(C.RD_KAFKA_OFFSET_BEGINNING)+
const OffsetBeginning = Offset(C.RD_KAFKA_OFFSET_BEGINNING)
OffsetEnd represents the latest offset (logical)
-const OffsetEnd = Offset(C.RD_KAFKA_OFFSET_END)+
const OffsetEnd = Offset(C.RD_KAFKA_OFFSET_END)
OffsetInvalid represents an invalid/unspecified offset
-const OffsetInvalid = Offset(C.RD_KAFKA_OFFSET_INVALID)+
const OffsetInvalid = Offset(C.RD_KAFKA_OFFSET_INVALID)
OffsetStored represents a stored offset
-const OffsetStored = Offset(C.RD_KAFKA_OFFSET_STORED)+
const OffsetStored = Offset(C.RD_KAFKA_OFFSET_STORED)
PartitionAny represents any partition (for partitioning), or unspecified value (for all other cases)
-const PartitionAny = int32(C.RD_KAFKA_PARTITION_UA)+
const PartitionAny = int32(C.RD_KAFKA_PARTITION_UA)
func LibraryVersion() (int, string)+
func LibraryVersion() (int, string)
LibraryVersion returns the underlying librdkafka library version as a (version_int, version_str) tuple.
func WriteErrorCodes(f *os.File)+
func WriteErrorCodes(f *os.File)
WriteErrorCodes writes Go error code constants to file from the librdkafka error codes. @@ -1591,7 +1591,7 @@
type ACLBindingFilter = ACLBinding
type ACLBindingFilters []ACLBindingFilter
type ACLBindings []ACLBinding
func (a ACLBindings) Len() int+
func (a ACLBindings) Len() int
func (a ACLBindings) Less(i, j int) bool+
func (a ACLBindings) Less(i, j int) bool
func (a ACLBindings) Swap(i, j int)+
func (a ACLBindings) Swap(i, j int)
ACLOperation enumerates the different types of ACL operation.
-type ACLOperation int+
type ACLOperation int
func ACLOperationFromString(aclOperationString string) (ACLOperation, error)+
func ACLOperationFromString(aclOperationString string) (ACLOperation, error)
ACLOperationFromString translates a ACL operation name to a ACLOperation value.
func (o ACLOperation) String() string+
func (o ACLOperation) String() string
String returns the human-readable representation of an ACLOperation
ACLPermissionType enumerates the different types of ACL permission types.
-type ACLPermissionType int+
type ACLPermissionType int
func ACLPermissionTypeFromString(aclPermissionTypeString string) (ACLPermissionType, error)+
func ACLPermissionTypeFromString(aclPermissionTypeString string) (ACLPermissionType, error)
ACLPermissionTypeFromString translates a ACL permission type name to a ACLPermissionType value.
func (o ACLPermissionType) String() string+
func (o ACLPermissionType) String() string
String returns the human-readable representation of an ACLPermissionType
func NewAdminClient(conf *ConfigMap) (*AdminClient, error)+
func NewAdminClient(conf *ConfigMap) (*AdminClient, error)
NewAdminClient creats a new AdminClient instance with a new underlying client instance
func NewAdminClientFromConsumer(c *Consumer) (a *AdminClient, err error)+
func NewAdminClientFromConsumer(c *Consumer) (a *AdminClient, err error)
NewAdminClientFromConsumer derives a new AdminClient from an existing Consumer instance. The AdminClient will use the same configuration and connections as the parent instance.
func NewAdminClientFromProducer(p *Producer) (a *AdminClient, err error)+
func NewAdminClientFromProducer(p *Producer) (a *AdminClient, err error)
NewAdminClientFromProducer derives a new AdminClient from an existing Producer instance. The AdminClient will use the same configuration and connections as the parent instance.
func (a *AdminClient) AlterConfigs(ctx context.Context, resources []ConfigResource, options ...AlterConfigsAdminOption) (result []ConfigResourceResult, err error)+
func (a *AdminClient) AlterConfigs(ctx context.Context, resources []ConfigResource, options ...AlterConfigsAdminOption) (result []ConfigResourceResult, err error)
AlterConfigs alters/updates cluster resource configuration.
@@ -1859,7 +1859,7 @@func (a *AdminClient) ClusterID(ctx context.Context) (clusterID string, err error)+
func (a *AdminClient) ClusterID(ctx context.Context) (clusterID string, err error)
ClusterID returns the cluster ID as reported in broker metadata.
@@ -1893,14 +1893,14 @@func (a *AdminClient) ControllerID(ctx context.Context) (controllerID int32, err error)+
func (a *AdminClient) ControllerID(ctx context.Context) (controllerID int32, err error)
ControllerID returns the broker ID of the current controller as reported in broker metadata. @@ -1915,14 +1915,14 @@
func (a *AdminClient) CreateACLs(ctx context.Context, aclBindings ACLBindings, options ...CreateACLsAdminOption) (result []CreateACLResult, err error)+
func (a *AdminClient) CreateACLs(ctx context.Context, aclBindings ACLBindings, options ...CreateACLsAdminOption) (result []CreateACLResult, err error)
CreateACLs creates one or more ACL bindings.
@@ -1939,27 +1939,27 @@func (a *AdminClient) CreatePartitions(ctx context.Context, partitions []PartitionsSpecification, options ...CreatePartitionsAdminOption) (result []TopicResult, err error)+
func (a *AdminClient) CreatePartitions(ctx context.Context, partitions []PartitionsSpecification, options ...CreatePartitionsAdminOption) (result []TopicResult, err error)
CreatePartitions creates additional partitions for topics.
func (a *AdminClient) CreateTopics(ctx context.Context, topics []TopicSpecification, options ...CreateTopicsAdminOption) (result []TopicResult, err error)+
func (a *AdminClient) CreateTopics(ctx context.Context, topics []TopicSpecification, options ...CreateTopicsAdminOption) (result []TopicResult, err error)
CreateTopics creates topics in cluster.
@@ -1975,14 +1975,14 @@func (a *AdminClient) DeleteACLs(ctx context.Context, aclBindingFilters ACLBindingFilters, options ...DeleteACLsAdminOption) (result []DeleteACLsResult, err error)+
func (a *AdminClient) DeleteACLs(ctx context.Context, aclBindingFilters ACLBindingFilters, options ...DeleteACLsAdminOption) (result []DeleteACLsResult, err error)
DeleteACLs deletes ACL bindings matching one or more ACL binding filters.
@@ -2004,14 +2004,14 @@func (a *AdminClient) DeleteTopics(ctx context.Context, topics []string, options ...DeleteTopicsAdminOption) (result []TopicResult, err error)+
func (a *AdminClient) DeleteTopics(ctx context.Context, topics []string, options ...DeleteTopicsAdminOption) (result []TopicResult, err error)
DeleteTopics deletes a batch of topics.
@@ -2027,14 +2027,14 @@func (a *AdminClient) DescribeACLs(ctx context.Context, aclBindingFilter ACLBindingFilter, options ...DescribeACLsAdminOption) (result *DescribeACLsResult, err error)+
func (a *AdminClient) DescribeACLs(ctx context.Context, aclBindingFilter ACLBindingFilter, options ...DescribeACLsAdminOption) (result *DescribeACLsResult, err error)
DescribeACLs matches ACL bindings by filter.
@@ -2056,14 +2056,14 @@func (a *AdminClient) DescribeConfigs(ctx context.Context, resources []ConfigResource, options ...DescribeConfigsAdminOption) (result []ConfigResourceResult, err error)+
func (a *AdminClient) DescribeConfigs(ctx context.Context, resources []ConfigResource, options ...DescribeConfigsAdminOption) (result []ConfigResourceResult, err error)
DescribeConfigs retrieves configuration for cluster resources.
@@ -2096,14 +2096,14 @@func (a *AdminClient) GetMetadata(topic *string, allTopics bool, timeoutMs int) (*Metadata, error)+
func (a *AdminClient) GetMetadata(topic *string, allTopics bool, timeoutMs int) (*Metadata, error)
GetMetadata queries broker for cluster and topic metadata. If topic is non-nil only information about that topic is returned, else if @@ -2113,14 +2113,14 @@
func (a *AdminClient) SetOAuthBearerToken(oauthBearerToken OAuthBearerToken) error+
func (a *AdminClient) SetOAuthBearerToken(oauthBearerToken OAuthBearerToken) error
SetOAuthBearerToken sets the the data to be transmitted to a broker during SASL/OAUTHBEARER authentication. It will return nil @@ -2138,14 +2138,14 @@
func (a *AdminClient) SetOAuthBearerTokenFailure(errstr string) error+
func (a *AdminClient) SetOAuthBearerTokenFailure(errstr string) error
SetOAuthBearerTokenFailure sets the error message describing why token retrieval/setting failed; it also schedules a new token refresh event for 10 @@ -2157,20 +2157,20 @@
func (a *AdminClient) String() string+
func (a *AdminClient) String() string
String returns a human readable name for an AdminClient instance
func SetAdminOperationTimeout(t time.Duration) (ao AdminOptionOperationTimeout)+
func SetAdminOperationTimeout(t time.Duration) (ao AdminOptionOperationTimeout)
SetAdminOperationTimeout sets the broker's operation timeout, such as the timeout for CreateTopics to complete the creation of topics on the controller @@ -2245,7 +2245,7 @@
func SetAdminRequestTimeout(t time.Duration) (ao AdminOptionRequestTimeout)+
func SetAdminRequestTimeout(t time.Duration) (ao AdminOptionRequestTimeout)
SetAdminRequestTimeout sets the overall request timeout, including broker lookup, request transmission, operation time on broker, and response. @@ -2288,7 +2288,7 @@
func SetAdminValidateOnly(validateOnly bool) (ao AdminOptionValidateOnly)+
func SetAdminValidateOnly(validateOnly bool) (ao AdminOptionValidateOnly)
SetAdminValidateOnly tells the broker to only validate the request, without performing the requested operation (create topics, etc). @@ -2331,7 +2331,7 @@
type AlterOperation int+
type AlterOperation int
func (o AlterOperation) String() string+
func (o AlterOperation) String() string
String returns the human-readable representation of an AlterOperation
func (e AssignedPartitions) String() string+
func (e AssignedPartitions) String() string
type BrokerMetadata struct {
- ID int32
- Host string
- Port int
+ ID int32
+ Host string
+ Port int
}
type ConfigEntry struct {
// Name of configuration entry, e.g., topic configuration property name.
- Name string
+ Name string
// Value of configuration entry.
- Value string
+ Value string
// Operation to perform on the entry.
Operation AlterOperation
}
func StringMapToConfigEntries(stringMap map[string]string, operation AlterOperation) []ConfigEntry+
func StringMapToConfigEntries(stringMap map[string]string, operation AlterOperation) []ConfigEntry
StringMapToConfigEntries creates a new map of ConfigEntry objects from the provided string map. The AlterOperation is set on each created entry.
func (c ConfigEntry) String() string+
func (c ConfigEntry) String() string
String returns a human-readable representation of a ConfigEntry.
type ConfigEntryResult struct {
// Name of configuration entry, e.g., topic configuration property name.
- Name string
+ Name string
// Value of configuration entry.
- Value string
+ Value string
// Source indicates the configuration source.
Source ConfigSource
// IsReadOnly indicates whether the configuration entry can be altered.
- IsReadOnly bool
+ IsReadOnly bool
// IsSensitive indicates whether the configuration entry contains sensitive information, in which case the value will be unset.
- IsSensitive bool
+ IsSensitive bool
// IsSynonym indicates whether the configuration entry is a synonym for another configuration property.
- IsSynonym bool
+ IsSynonym bool
// Synonyms contains a map of configuration entries that are synonyms to this configuration entry.
- Synonyms map[string]ConfigEntryResult
+ Synonyms map[string]ConfigEntryResult
}
func (c ConfigEntryResult) String() string+
func (c ConfigEntryResult) String() string
String returns a human-readable representation of a ConfigEntryResult.
type ConfigMap map[string]ConfigValue+
type ConfigMap map[string]ConfigValue
func (m ConfigMap) Get(key string, defval ConfigValue) (ConfigValue, error)+
func (m ConfigMap) Get(key string, defval ConfigValue) (ConfigValue, error)
Get finds the given key in the ConfigMap and returns its value. If the key is not found `defval` is returned. @@ -2553,28 +2553,28 @@
func (m ConfigMap) Set(kv string) error+
func (m ConfigMap) Set(kv string) error
Set implements flag.Set (command line argument parser) as a convenience for `-X key=value` config.
func (m ConfigMap) SetKey(key string, value ConfigValue) error+
func (m ConfigMap) SetKey(key string, value ConfigValue) error
SetKey sets configuration property key to value.
@@ -2584,7 +2584,7 @@func (c ConfigResource) String() string+
func (c ConfigResource) String() string
String returns a human-readable representation of a ConfigResource
func (c ConfigResourceResult) String() string+
func (c ConfigResourceResult) String() string
String returns a human-readable representation of a ConfigResourceResult.
ConfigSource represents an Apache Kafka config source
-type ConfigSource int+
type ConfigSource int
func (t ConfigSource) String() string+
func (t ConfigSource) String() string
String returns the human-readable representation of a ConfigSource type
type ConfigValue interface{}
func NewConsumer(conf *ConfigMap) (*Consumer, error)+
func NewConsumer(conf *ConfigMap) (*Consumer, error)
NewConsumer creates a new high-level Consumer instance.
@@ -2751,14 +2751,14 @@func (c *Consumer) Assign(partitions []TopicPartition) (err error)+
func (c *Consumer) Assign(partitions []TopicPartition) (err error)
Assign an atomic set of partitions to consume.
@@ -2774,27 +2774,27 @@func (c *Consumer) Assignment() (partitions []TopicPartition, err error)+
func (c *Consumer) Assignment() (partitions []TopicPartition, err error)
Assignment returns the current partition assignments
func (c *Consumer) AssignmentLost() bool+
func (c *Consumer) AssignmentLost() bool
AssignmentLost returns true if current partition assignment has been lost. This method is only applicable for use with a subscribing consumer when @@ -2804,28 +2804,28 @@
func (c *Consumer) Close() (err error)+
func (c *Consumer) Close() (err error)
Close Consumer instance. The object is no longer usable after this call.
func (c *Consumer) Commit() ([]TopicPartition, error)+
func (c *Consumer) Commit() ([]TopicPartition, error)
Commit offsets for currently assigned partitions This is a blocking call. @@ -2833,14 +2833,14 @@
func (c *Consumer) CommitMessage(m *Message) ([]TopicPartition, error)+
func (c *Consumer) CommitMessage(m *Message) ([]TopicPartition, error)
CommitMessage commits offset based on the provided message. This is a blocking call. @@ -2848,14 +2848,14 @@
func (c *Consumer) CommitOffsets(offsets []TopicPartition) ([]TopicPartition, error)+
func (c *Consumer) CommitOffsets(offsets []TopicPartition) ([]TopicPartition, error)
CommitOffsets commits the provided list of offsets This is a blocking call. @@ -2863,20 +2863,20 @@
func (c *Consumer) Committed(partitions []TopicPartition, timeoutMs int) (offsets []TopicPartition, err error)+
func (c *Consumer) Committed(partitions []TopicPartition, timeoutMs int) (offsets []TopicPartition, err error)
Committed retrieves committed offsets for the given set of partitions
func (c *Consumer) GetConsumerGroupMetadata() (*ConsumerGroupMetadata, error)+
func (c *Consumer) GetConsumerGroupMetadata() (*ConsumerGroupMetadata, error)
GetConsumerGroupMetadata returns the consumer's current group metadata. This object should be passed to the transactional producer's @@ -2904,14 +2904,14 @@
func (c *Consumer) GetMetadata(topic *string, allTopics bool, timeoutMs int) (*Metadata, error)+
func (c *Consumer) GetMetadata(topic *string, allTopics bool, timeoutMs int) (*Metadata, error)
GetMetadata queries broker for cluster and topic metadata. If topic is non-nil only information about that topic is returned, else if @@ -2921,14 +2921,14 @@
func (c *Consumer) GetRebalanceProtocol() string+
func (c *Consumer) GetRebalanceProtocol() string
GetRebalanceProtocol returns the current consumer group rebalance protocol, which is either "EAGER" or "COOPERATIVE". @@ -2938,14 +2938,14 @@
func (c *Consumer) GetWatermarkOffsets(topic string, partition int32) (low, high int64, err error)+
func (c *Consumer) GetWatermarkOffsets(topic string, partition int32) (low, high int64, err error)
GetWatermarkOffsets returns the cached low and high offsets for the given topic and partition. The high offset is populated on every fetch response or via calling QueryWatermarkOffsets. @@ -2954,14 +2954,14 @@
func (c *Consumer) IncrementalAssign(partitions []TopicPartition) (err error)+
func (c *Consumer) IncrementalAssign(partitions []TopicPartition) (err error)
IncrementalAssign adds the specified partitions to the current set of partitions to consume. @@ -2978,14 +2978,14 @@
func (c *Consumer) IncrementalUnassign(partitions []TopicPartition) (err error)+
func (c *Consumer) IncrementalUnassign(partitions []TopicPartition) (err error)
IncrementalUnassign removes the specified partitions from the current set of partitions to consume. @@ -2998,7 +2998,7 @@
func (c *Consumer) OffsetsForTimes(times []TopicPartition, timeoutMs int) (offsets []TopicPartition, err error)+
func (c *Consumer) OffsetsForTimes(times []TopicPartition, timeoutMs int) (offsets []TopicPartition, err error)
OffsetsForTimes looks up offsets by timestamp for the given partitions.
@@ -3042,14 +3042,14 @@func (c *Consumer) Pause(partitions []TopicPartition) (err error)+
func (c *Consumer) Pause(partitions []TopicPartition) (err error)
Pause consumption for the provided list of partitions
@@ -3060,14 +3060,14 @@func (c *Consumer) Poll(timeoutMs int) (event Event)+
func (c *Consumer) Poll(timeoutMs int) (event Event)
Poll the consumer for messages or events.
@@ -3084,14 +3084,14 @@func (c *Consumer) Position(partitions []TopicPartition) (offsets []TopicPartition, err error)+
func (c *Consumer) Position(partitions []TopicPartition) (offsets []TopicPartition, err error)
Position returns the current consume position for the given partitions. Typical use is to call Assignment() to get the partition list @@ -3102,27 +3102,27 @@
func (c *Consumer) QueryWatermarkOffsets(topic string, partition int32, timeoutMs int) (low, high int64, err error)+
func (c *Consumer) QueryWatermarkOffsets(topic string, partition int32, timeoutMs int) (low, high int64, err error)
QueryWatermarkOffsets queries the broker for the low and high offsets for the given topic and partition.
func (c *Consumer) ReadMessage(timeout time.Duration) (*Message, error)+
func (c *Consumer) ReadMessage(timeout time.Duration) (*Message, error)
ReadMessage polls the consumer for a message.
@@ -3149,27 +3149,27 @@func (c *Consumer) Resume(partitions []TopicPartition) (err error)+
func (c *Consumer) Resume(partitions []TopicPartition) (err error)
Resume consumption for the provided list of partitions
func (c *Consumer) Seek(partition TopicPartition, timeoutMs int) error+
func (c *Consumer) Seek(partition TopicPartition, timeoutMs int) error
Seek seeks the given topic partitions using the offset from the TopicPartition.
@@ -3191,14 +3191,14 @@func (c *Consumer) SetOAuthBearerToken(oauthBearerToken OAuthBearerToken) error+
func (c *Consumer) SetOAuthBearerToken(oauthBearerToken OAuthBearerToken) error
SetOAuthBearerToken sets the the data to be transmitted to a broker during SASL/OAUTHBEARER authentication. It will return nil @@ -3216,14 +3216,14 @@
func (c *Consumer) SetOAuthBearerTokenFailure(errstr string) error+
func (c *Consumer) SetOAuthBearerTokenFailure(errstr string) error
SetOAuthBearerTokenFailure sets the error message describing why token retrieval/setting failed; it also schedules a new token refresh event for 10 @@ -3235,28 +3235,28 @@
func (c *Consumer) StoreMessage(m *Message) (storedOffsets []TopicPartition, err error)+
func (c *Consumer) StoreMessage(m *Message) (storedOffsets []TopicPartition, err error)
StoreMessage stores offset based on the provided message. This is a convenience method that uses StoreOffsets to do the actual work.
func (c *Consumer) StoreOffsets(offsets []TopicPartition) (storedOffsets []TopicPartition, err error)+
func (c *Consumer) StoreOffsets(offsets []TopicPartition) (storedOffsets []TopicPartition, err error)
StoreOffsets stores the provided list of offsets that will be committed to the offset store according to `auto.commit.interval.ms` or manual @@ -3269,87 +3269,87 @@
func (c *Consumer) String() string+
func (c *Consumer) String() string
Strings returns a human readable name for a Consumer instance
func (c *Consumer) Subscribe(topic string, rebalanceCb RebalanceCb) error+
func (c *Consumer) Subscribe(topic string, rebalanceCb RebalanceCb) error
Subscribe to a single topic This replaces the current subscription
func (c *Consumer) SubscribeTopics(topics []string, rebalanceCb RebalanceCb) (err error)+
func (c *Consumer) SubscribeTopics(topics []string, rebalanceCb RebalanceCb) (err error)
SubscribeTopics subscribes to the provided list of topics. This replaces the current subscription.
func (c *Consumer) Subscription() (topics []string, err error)+
func (c *Consumer) Subscription() (topics []string, err error)
Subscription returns the current subscription as set by Subscribe()
func (c *Consumer) Unassign() (err error)+
func (c *Consumer) Unassign() (err error)
Unassign the current set of partitions to consume.
func (c *Consumer) Unsubscribe() (err error)+
func (c *Consumer) Unsubscribe() (err error)
Unsubscribe from the current subscription, if any.
func NewTestConsumerGroupMetadata(groupID string) (*ConsumerGroupMetadata, error)+
func NewTestConsumerGroupMetadata(groupID string) (*ConsumerGroupMetadata, error)
NewTestConsumerGroupMetadata creates a new consumer group metadata instance mainly for testing use. @@ -3380,7 +3380,7 @@
type DeleteACLsResult = DescribeACLsResult
func NewError(code ErrorCode, str string, fatal bool) (err Error)+
func NewError(code ErrorCode, str string, fatal bool) (err Error)
NewError creates a new Error.
func (e Error) Error() string+
func (e Error) Error() string
Error returns a human readable representation of an Error Same as Error.String()
func (e Error) IsFatal() bool+
func (e Error) IsFatal() bool
IsFatal returns true if the error is a fatal error. A fatal error indicates the client instance is no longer operable and @@ -3627,14 +3627,14 @@
func (e Error) IsRetriable() bool+
func (e Error) IsRetriable() bool
IsRetriable returns true if the operation that caused this error may be retried. @@ -3642,27 +3642,27 @@
func (e Error) String() string+
func (e Error) String() string
String returns a human readable representation of an Error
func (e Error) TxnRequiresAbort() bool+
func (e Error) TxnRequiresAbort() bool
TxnRequiresAbort returns true if the error is an abortable transaction error that requires the application to abort the current transaction with @@ -3672,7 +3672,7 @@
ErrorCode is the integer representation of local and broker error codes
-type ErrorCode int+
type ErrorCode int
const (
// ErrBadMsg Local: Bad message format
- ErrBadMsg ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__BAD_MSG)
+ ErrBadMsg ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__BAD_MSG)
// ErrBadCompression Local: Invalid compressed data
- ErrBadCompression ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__BAD_COMPRESSION)
+ ErrBadCompression ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__BAD_COMPRESSION)
// ErrDestroy Local: Broker handle destroyed
- ErrDestroy ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__DESTROY)
+ ErrDestroy ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__DESTROY)
// ErrFail Local: Communication failure with broker
- ErrFail ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__FAIL)
+ ErrFail ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__FAIL)
// ErrTransport Local: Broker transport failure
- ErrTransport ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__TRANSPORT)
+ ErrTransport ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__TRANSPORT)
// ErrCritSysResource Local: Critical system resource failure
- ErrCritSysResource ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__CRIT_SYS_RESOURCE)
+ ErrCritSysResource ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__CRIT_SYS_RESOURCE)
// ErrResolve Local: Host resolution failure
- ErrResolve ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__RESOLVE)
+ ErrResolve ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__RESOLVE)
// ErrMsgTimedOut Local: Message timed out
- ErrMsgTimedOut ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__MSG_TIMED_OUT)
+ ErrMsgTimedOut ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__MSG_TIMED_OUT)
// ErrPartitionEOF Broker: No more messages
- ErrPartitionEOF ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__PARTITION_EOF)
+ ErrPartitionEOF ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__PARTITION_EOF)
// ErrUnknownPartition Local: Unknown partition
- ErrUnknownPartition ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__UNKNOWN_PARTITION)
+ ErrUnknownPartition ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__UNKNOWN_PARTITION)
// ErrFs Local: File or filesystem error
- ErrFs ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__FS)
+ ErrFs ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__FS)
// ErrUnknownTopic Local: Unknown topic
- ErrUnknownTopic ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__UNKNOWN_TOPIC)
+ ErrUnknownTopic ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__UNKNOWN_TOPIC)
// ErrAllBrokersDown Local: All broker connections are down
- ErrAllBrokersDown ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__ALL_BROKERS_DOWN)
+ ErrAllBrokersDown ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__ALL_BROKERS_DOWN)
// ErrInvalidArg Local: Invalid argument or configuration
- ErrInvalidArg ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__INVALID_ARG)
+ ErrInvalidArg ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__INVALID_ARG)
// ErrTimedOut Local: Timed out
- ErrTimedOut ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__TIMED_OUT)
+ ErrTimedOut ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__TIMED_OUT)
// ErrQueueFull Local: Queue full
- ErrQueueFull ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__QUEUE_FULL)
+ ErrQueueFull ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__QUEUE_FULL)
// ErrIsrInsuff Local: ISR count insufficient
- ErrIsrInsuff ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__ISR_INSUFF)
+ ErrIsrInsuff ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__ISR_INSUFF)
// ErrNodeUpdate Local: Broker node update
- ErrNodeUpdate ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__NODE_UPDATE)
+ ErrNodeUpdate ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__NODE_UPDATE)
// ErrSsl Local: SSL error
- ErrSsl ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__SSL)
+ ErrSsl ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__SSL)
// ErrWaitCoord Local: Waiting for coordinator
- ErrWaitCoord ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__WAIT_COORD)
+ ErrWaitCoord ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__WAIT_COORD)
// ErrUnknownGroup Local: Unknown group
- ErrUnknownGroup ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__UNKNOWN_GROUP)
+ ErrUnknownGroup ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__UNKNOWN_GROUP)
// ErrInProgress Local: Operation in progress
- ErrInProgress ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__IN_PROGRESS)
+ ErrInProgress ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__IN_PROGRESS)
// ErrPrevInProgress Local: Previous operation in progress
- ErrPrevInProgress ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__PREV_IN_PROGRESS)
+ ErrPrevInProgress ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__PREV_IN_PROGRESS)
// ErrExistingSubscription Local: Existing subscription
- ErrExistingSubscription ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__EXISTING_SUBSCRIPTION)
+ ErrExistingSubscription ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__EXISTING_SUBSCRIPTION)
// ErrAssignPartitions Local: Assign partitions
- ErrAssignPartitions ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS)
+ ErrAssignPartitions ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS)
// ErrRevokePartitions Local: Revoke partitions
- ErrRevokePartitions ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS)
+ ErrRevokePartitions ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS)
// ErrConflict Local: Conflicting use
- ErrConflict ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__CONFLICT)
+ ErrConflict ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__CONFLICT)
// ErrState Local: Erroneous state
- ErrState ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__STATE)
+ ErrState ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__STATE)
// ErrUnknownProtocol Local: Unknown protocol
- ErrUnknownProtocol ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__UNKNOWN_PROTOCOL)
+ ErrUnknownProtocol ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__UNKNOWN_PROTOCOL)
// ErrNotImplemented Local: Not implemented
- ErrNotImplemented ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__NOT_IMPLEMENTED)
+ ErrNotImplemented ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__NOT_IMPLEMENTED)
// ErrAuthentication Local: Authentication failure
- ErrAuthentication ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__AUTHENTICATION)
+ ErrAuthentication ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__AUTHENTICATION)
// ErrNoOffset Local: No offset stored
- ErrNoOffset ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__NO_OFFSET)
+ ErrNoOffset ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__NO_OFFSET)
// ErrOutdated Local: Outdated
- ErrOutdated ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__OUTDATED)
+ ErrOutdated ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__OUTDATED)
// ErrTimedOutQueue Local: Timed out in queue
- ErrTimedOutQueue ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__TIMED_OUT_QUEUE)
+ ErrTimedOutQueue ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__TIMED_OUT_QUEUE)
// ErrUnsupportedFeature Local: Required feature not supported by broker
- ErrUnsupportedFeature ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE)
+ ErrUnsupportedFeature ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__UNSUPPORTED_FEATURE)
// ErrWaitCache Local: Awaiting cache update
- ErrWaitCache ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__WAIT_CACHE)
+ ErrWaitCache ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__WAIT_CACHE)
// ErrIntr Local: Operation interrupted
- ErrIntr ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__INTR)
+ ErrIntr ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__INTR)
// ErrKeySerialization Local: Key serialization error
- ErrKeySerialization ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__KEY_SERIALIZATION)
+ ErrKeySerialization ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__KEY_SERIALIZATION)
// ErrValueSerialization Local: Value serialization error
- ErrValueSerialization ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__VALUE_SERIALIZATION)
+ ErrValueSerialization ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__VALUE_SERIALIZATION)
// ErrKeyDeserialization Local: Key deserialization error
- ErrKeyDeserialization ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__KEY_DESERIALIZATION)
+ ErrKeyDeserialization ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__KEY_DESERIALIZATION)
// ErrValueDeserialization Local: Value deserialization error
- ErrValueDeserialization ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__VALUE_DESERIALIZATION)
+ ErrValueDeserialization ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__VALUE_DESERIALIZATION)
// ErrPartial Local: Partial response
- ErrPartial ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__PARTIAL)
+ ErrPartial ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__PARTIAL)
// ErrReadOnly Local: Read-only object
- ErrReadOnly ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__READ_ONLY)
+ ErrReadOnly ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__READ_ONLY)
// ErrNoent Local: No such entry
- ErrNoent ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__NOENT)
+ ErrNoent ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__NOENT)
// ErrUnderflow Local: Read underflow
- ErrUnderflow ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__UNDERFLOW)
+ ErrUnderflow ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__UNDERFLOW)
// ErrInvalidType Local: Invalid type
- ErrInvalidType ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__INVALID_TYPE)
+ ErrInvalidType ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__INVALID_TYPE)
// ErrRetry Local: Retry operation
- ErrRetry ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__RETRY)
+ ErrRetry ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__RETRY)
// ErrPurgeQueue Local: Purged in queue
- ErrPurgeQueue ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__PURGE_QUEUE)
+ ErrPurgeQueue ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__PURGE_QUEUE)
// ErrPurgeInflight Local: Purged in flight
- ErrPurgeInflight ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__PURGE_INFLIGHT)
+ ErrPurgeInflight ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__PURGE_INFLIGHT)
// ErrFatal Local: Fatal error
- ErrFatal ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__FATAL)
+ ErrFatal ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__FATAL)
// ErrInconsistent Local: Inconsistent state
- ErrInconsistent ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__INCONSISTENT)
+ ErrInconsistent ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__INCONSISTENT)
// ErrGaplessGuarantee Local: Gap-less ordering would not be guaranteed if proceeding
- ErrGaplessGuarantee ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__GAPLESS_GUARANTEE)
+ ErrGaplessGuarantee ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__GAPLESS_GUARANTEE)
// ErrMaxPollExceeded Local: Maximum application poll interval (max.poll.interval.ms) exceeded
- ErrMaxPollExceeded ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__MAX_POLL_EXCEEDED)
+ ErrMaxPollExceeded ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__MAX_POLL_EXCEEDED)
// ErrUnknownBroker Local: Unknown broker
- ErrUnknownBroker ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__UNKNOWN_BROKER)
+ ErrUnknownBroker ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__UNKNOWN_BROKER)
// ErrNotConfigured Local: Functionality not configured
- ErrNotConfigured ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__NOT_CONFIGURED)
+ ErrNotConfigured ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__NOT_CONFIGURED)
// ErrFenced Local: This instance has been fenced by a newer instance
- ErrFenced ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__FENCED)
+ ErrFenced ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__FENCED)
// ErrApplication Local: Application generated error
- ErrApplication ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__APPLICATION)
+ ErrApplication ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__APPLICATION)
// ErrAssignmentLost Local: Group partition assignment lost
- ErrAssignmentLost ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__ASSIGNMENT_LOST)
+ ErrAssignmentLost ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__ASSIGNMENT_LOST)
// ErrNoop Local: No operation performed
- ErrNoop ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__NOOP)
+ ErrNoop ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__NOOP)
// ErrAutoOffsetReset Local: No offset to automatically reset to
- ErrAutoOffsetReset ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__AUTO_OFFSET_RESET)
+ ErrAutoOffsetReset ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR__AUTO_OFFSET_RESET)
// ErrUnknown Unknown broker error
- ErrUnknown ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_UNKNOWN)
+ ErrUnknown ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_UNKNOWN)
// ErrNoError Success
- ErrNoError ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_NO_ERROR)
+ ErrNoError ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_NO_ERROR)
// ErrOffsetOutOfRange Broker: Offset out of range
- ErrOffsetOutOfRange ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_OFFSET_OUT_OF_RANGE)
+ ErrOffsetOutOfRange ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_OFFSET_OUT_OF_RANGE)
// ErrInvalidMsg Broker: Invalid message
- ErrInvalidMsg ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INVALID_MSG)
+ ErrInvalidMsg ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INVALID_MSG)
// ErrUnknownTopicOrPart Broker: Unknown topic or partition
- ErrUnknownTopicOrPart ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_UNKNOWN_TOPIC_OR_PART)
+ ErrUnknownTopicOrPart ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_UNKNOWN_TOPIC_OR_PART)
// ErrInvalidMsgSize Broker: Invalid message size
- ErrInvalidMsgSize ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INVALID_MSG_SIZE)
+ ErrInvalidMsgSize ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INVALID_MSG_SIZE)
// ErrLeaderNotAvailable Broker: Leader not available
- ErrLeaderNotAvailable ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_LEADER_NOT_AVAILABLE)
+ ErrLeaderNotAvailable ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_LEADER_NOT_AVAILABLE)
// ErrNotLeaderForPartition Broker: Not leader for partition
- ErrNotLeaderForPartition ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_NOT_LEADER_FOR_PARTITION)
+ ErrNotLeaderForPartition ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_NOT_LEADER_FOR_PARTITION)
// ErrRequestTimedOut Broker: Request timed out
- ErrRequestTimedOut ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_REQUEST_TIMED_OUT)
+ ErrRequestTimedOut ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_REQUEST_TIMED_OUT)
// ErrBrokerNotAvailable Broker: Broker not available
- ErrBrokerNotAvailable ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_BROKER_NOT_AVAILABLE)
+ ErrBrokerNotAvailable ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_BROKER_NOT_AVAILABLE)
// ErrReplicaNotAvailable Broker: Replica not available
- ErrReplicaNotAvailable ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_REPLICA_NOT_AVAILABLE)
+ ErrReplicaNotAvailable ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_REPLICA_NOT_AVAILABLE)
// ErrMsgSizeTooLarge Broker: Message size too large
- ErrMsgSizeTooLarge ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_MSG_SIZE_TOO_LARGE)
+ ErrMsgSizeTooLarge ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_MSG_SIZE_TOO_LARGE)
// ErrStaleCtrlEpoch Broker: StaleControllerEpochCode
- ErrStaleCtrlEpoch ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_STALE_CTRL_EPOCH)
+ ErrStaleCtrlEpoch ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_STALE_CTRL_EPOCH)
// ErrOffsetMetadataTooLarge Broker: Offset metadata string too large
- ErrOffsetMetadataTooLarge ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_OFFSET_METADATA_TOO_LARGE)
+ ErrOffsetMetadataTooLarge ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_OFFSET_METADATA_TOO_LARGE)
// ErrNetworkException Broker: Broker disconnected before response received
- ErrNetworkException ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_NETWORK_EXCEPTION)
+ ErrNetworkException ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_NETWORK_EXCEPTION)
// ErrCoordinatorLoadInProgress Broker: Coordinator load in progress
- ErrCoordinatorLoadInProgress ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_COORDINATOR_LOAD_IN_PROGRESS)
+ ErrCoordinatorLoadInProgress ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_COORDINATOR_LOAD_IN_PROGRESS)
// ErrCoordinatorNotAvailable Broker: Coordinator not available
- ErrCoordinatorNotAvailable ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_COORDINATOR_NOT_AVAILABLE)
+ ErrCoordinatorNotAvailable ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_COORDINATOR_NOT_AVAILABLE)
// ErrNotCoordinator Broker: Not coordinator
- ErrNotCoordinator ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_NOT_COORDINATOR)
+ ErrNotCoordinator ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_NOT_COORDINATOR)
// ErrTopicException Broker: Invalid topic
- ErrTopicException ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_TOPIC_EXCEPTION)
+ ErrTopicException ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_TOPIC_EXCEPTION)
// ErrRecordListTooLarge Broker: Message batch larger than configured server segment size
- ErrRecordListTooLarge ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_RECORD_LIST_TOO_LARGE)
+ ErrRecordListTooLarge ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_RECORD_LIST_TOO_LARGE)
// ErrNotEnoughReplicas Broker: Not enough in-sync replicas
- ErrNotEnoughReplicas ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_NOT_ENOUGH_REPLICAS)
+ ErrNotEnoughReplicas ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_NOT_ENOUGH_REPLICAS)
// ErrNotEnoughReplicasAfterAppend Broker: Message(s) written to insufficient number of in-sync replicas
- ErrNotEnoughReplicasAfterAppend ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_NOT_ENOUGH_REPLICAS_AFTER_APPEND)
+ ErrNotEnoughReplicasAfterAppend ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_NOT_ENOUGH_REPLICAS_AFTER_APPEND)
// ErrInvalidRequiredAcks Broker: Invalid required acks value
- ErrInvalidRequiredAcks ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INVALID_REQUIRED_ACKS)
+ ErrInvalidRequiredAcks ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INVALID_REQUIRED_ACKS)
// ErrIllegalGeneration Broker: Specified group generation id is not valid
- ErrIllegalGeneration ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_ILLEGAL_GENERATION)
+ ErrIllegalGeneration ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_ILLEGAL_GENERATION)
// ErrInconsistentGroupProtocol Broker: Inconsistent group protocol
- ErrInconsistentGroupProtocol ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INCONSISTENT_GROUP_PROTOCOL)
+ ErrInconsistentGroupProtocol ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INCONSISTENT_GROUP_PROTOCOL)
// ErrInvalidGroupID Broker: Invalid group.id
- ErrInvalidGroupID ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INVALID_GROUP_ID)
+ ErrInvalidGroupID ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INVALID_GROUP_ID)
// ErrUnknownMemberID Broker: Unknown member
- ErrUnknownMemberID ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_UNKNOWN_MEMBER_ID)
+ ErrUnknownMemberID ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_UNKNOWN_MEMBER_ID)
// ErrInvalidSessionTimeout Broker: Invalid session timeout
- ErrInvalidSessionTimeout ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INVALID_SESSION_TIMEOUT)
+ ErrInvalidSessionTimeout ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INVALID_SESSION_TIMEOUT)
// ErrRebalanceInProgress Broker: Group rebalance in progress
- ErrRebalanceInProgress ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_REBALANCE_IN_PROGRESS)
+ ErrRebalanceInProgress ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_REBALANCE_IN_PROGRESS)
// ErrInvalidCommitOffsetSize Broker: Commit offset data size is not valid
- ErrInvalidCommitOffsetSize ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INVALID_COMMIT_OFFSET_SIZE)
+ ErrInvalidCommitOffsetSize ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INVALID_COMMIT_OFFSET_SIZE)
// ErrTopicAuthorizationFailed Broker: Topic authorization failed
- ErrTopicAuthorizationFailed ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_TOPIC_AUTHORIZATION_FAILED)
+ ErrTopicAuthorizationFailed ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_TOPIC_AUTHORIZATION_FAILED)
// ErrGroupAuthorizationFailed Broker: Group authorization failed
- ErrGroupAuthorizationFailed ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_GROUP_AUTHORIZATION_FAILED)
+ ErrGroupAuthorizationFailed ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_GROUP_AUTHORIZATION_FAILED)
// ErrClusterAuthorizationFailed Broker: Cluster authorization failed
- ErrClusterAuthorizationFailed ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_CLUSTER_AUTHORIZATION_FAILED)
+ ErrClusterAuthorizationFailed ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_CLUSTER_AUTHORIZATION_FAILED)
// ErrInvalidTimestamp Broker: Invalid timestamp
- ErrInvalidTimestamp ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INVALID_TIMESTAMP)
+ ErrInvalidTimestamp ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INVALID_TIMESTAMP)
// ErrUnsupportedSaslMechanism Broker: Unsupported SASL mechanism
- ErrUnsupportedSaslMechanism ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_UNSUPPORTED_SASL_MECHANISM)
+ ErrUnsupportedSaslMechanism ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_UNSUPPORTED_SASL_MECHANISM)
// ErrIllegalSaslState Broker: Request not valid in current SASL state
- ErrIllegalSaslState ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_ILLEGAL_SASL_STATE)
+ ErrIllegalSaslState ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_ILLEGAL_SASL_STATE)
// ErrUnsupportedVersion Broker: API version not supported
- ErrUnsupportedVersion ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_UNSUPPORTED_VERSION)
+ ErrUnsupportedVersion ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_UNSUPPORTED_VERSION)
// ErrTopicAlreadyExists Broker: Topic already exists
- ErrTopicAlreadyExists ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_TOPIC_ALREADY_EXISTS)
+ ErrTopicAlreadyExists ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_TOPIC_ALREADY_EXISTS)
// ErrInvalidPartitions Broker: Invalid number of partitions
- ErrInvalidPartitions ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INVALID_PARTITIONS)
+ ErrInvalidPartitions ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INVALID_PARTITIONS)
// ErrInvalidReplicationFactor Broker: Invalid replication factor
- ErrInvalidReplicationFactor ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INVALID_REPLICATION_FACTOR)
+ ErrInvalidReplicationFactor ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INVALID_REPLICATION_FACTOR)
// ErrInvalidReplicaAssignment Broker: Invalid replica assignment
- ErrInvalidReplicaAssignment ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INVALID_REPLICA_ASSIGNMENT)
+ ErrInvalidReplicaAssignment ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INVALID_REPLICA_ASSIGNMENT)
// ErrInvalidConfig Broker: Configuration is invalid
- ErrInvalidConfig ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INVALID_CONFIG)
+ ErrInvalidConfig ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INVALID_CONFIG)
// ErrNotController Broker: Not controller for cluster
- ErrNotController ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_NOT_CONTROLLER)
+ ErrNotController ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_NOT_CONTROLLER)
// ErrInvalidRequest Broker: Invalid request
- ErrInvalidRequest ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INVALID_REQUEST)
+ ErrInvalidRequest ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INVALID_REQUEST)
// ErrUnsupportedForMessageFormat Broker: Message format on broker does not support request
- ErrUnsupportedForMessageFormat ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_UNSUPPORTED_FOR_MESSAGE_FORMAT)
+ ErrUnsupportedForMessageFormat ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_UNSUPPORTED_FOR_MESSAGE_FORMAT)
// ErrPolicyViolation Broker: Policy violation
- ErrPolicyViolation ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_POLICY_VIOLATION)
+ ErrPolicyViolation ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_POLICY_VIOLATION)
// ErrOutOfOrderSequenceNumber Broker: Broker received an out of order sequence number
- ErrOutOfOrderSequenceNumber ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_OUT_OF_ORDER_SEQUENCE_NUMBER)
+ ErrOutOfOrderSequenceNumber ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_OUT_OF_ORDER_SEQUENCE_NUMBER)
// ErrDuplicateSequenceNumber Broker: Broker received a duplicate sequence number
- ErrDuplicateSequenceNumber ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_DUPLICATE_SEQUENCE_NUMBER)
+ ErrDuplicateSequenceNumber ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_DUPLICATE_SEQUENCE_NUMBER)
// ErrInvalidProducerEpoch Broker: Producer attempted an operation with an old epoch
- ErrInvalidProducerEpoch ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INVALID_PRODUCER_EPOCH)
+ ErrInvalidProducerEpoch ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INVALID_PRODUCER_EPOCH)
// ErrInvalidTxnState Broker: Producer attempted a transactional operation in an invalid state
- ErrInvalidTxnState ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INVALID_TXN_STATE)
+ ErrInvalidTxnState ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INVALID_TXN_STATE)
// ErrInvalidProducerIDMapping Broker: Producer attempted to use a producer id which is not currently assigned to its transactional id
- ErrInvalidProducerIDMapping ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INVALID_PRODUCER_ID_MAPPING)
+ ErrInvalidProducerIDMapping ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INVALID_PRODUCER_ID_MAPPING)
// ErrInvalidTransactionTimeout Broker: Transaction timeout is larger than the maximum value allowed by the broker's max.transaction.timeout.ms
- ErrInvalidTransactionTimeout ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INVALID_TRANSACTION_TIMEOUT)
+ ErrInvalidTransactionTimeout ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INVALID_TRANSACTION_TIMEOUT)
// ErrConcurrentTransactions Broker: Producer attempted to update a transaction while another concurrent operation on the same transaction was ongoing
- ErrConcurrentTransactions ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_CONCURRENT_TRANSACTIONS)
+ ErrConcurrentTransactions ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_CONCURRENT_TRANSACTIONS)
// ErrTransactionCoordinatorFenced Broker: Indicates that the transaction coordinator sending a WriteTxnMarker is no longer the current coordinator for a given producer
- ErrTransactionCoordinatorFenced ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_TRANSACTION_COORDINATOR_FENCED)
+ ErrTransactionCoordinatorFenced ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_TRANSACTION_COORDINATOR_FENCED)
// ErrTransactionalIDAuthorizationFailed Broker: Transactional Id authorization failed
- ErrTransactionalIDAuthorizationFailed ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_TRANSACTIONAL_ID_AUTHORIZATION_FAILED)
+ ErrTransactionalIDAuthorizationFailed ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_TRANSACTIONAL_ID_AUTHORIZATION_FAILED)
// ErrSecurityDisabled Broker: Security features are disabled
- ErrSecurityDisabled ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_SECURITY_DISABLED)
+ ErrSecurityDisabled ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_SECURITY_DISABLED)
// ErrOperationNotAttempted Broker: Operation not attempted
- ErrOperationNotAttempted ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_OPERATION_NOT_ATTEMPTED)
+ ErrOperationNotAttempted ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_OPERATION_NOT_ATTEMPTED)
// ErrKafkaStorageError Broker: Disk error when trying to access log file on disk
- ErrKafkaStorageError ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_KAFKA_STORAGE_ERROR)
+ ErrKafkaStorageError ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_KAFKA_STORAGE_ERROR)
// ErrLogDirNotFound Broker: The user-specified log directory is not found in the broker config
- ErrLogDirNotFound ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_LOG_DIR_NOT_FOUND)
+ ErrLogDirNotFound ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_LOG_DIR_NOT_FOUND)
// ErrSaslAuthenticationFailed Broker: SASL Authentication failed
- ErrSaslAuthenticationFailed ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_SASL_AUTHENTICATION_FAILED)
+ ErrSaslAuthenticationFailed ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_SASL_AUTHENTICATION_FAILED)
// ErrUnknownProducerID Broker: Unknown Producer Id
- ErrUnknownProducerID ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_UNKNOWN_PRODUCER_ID)
+ ErrUnknownProducerID ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_UNKNOWN_PRODUCER_ID)
// ErrReassignmentInProgress Broker: Partition reassignment is in progress
- ErrReassignmentInProgress ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_REASSIGNMENT_IN_PROGRESS)
+ ErrReassignmentInProgress ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_REASSIGNMENT_IN_PROGRESS)
// ErrDelegationTokenAuthDisabled Broker: Delegation Token feature is not enabled
- ErrDelegationTokenAuthDisabled ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_DELEGATION_TOKEN_AUTH_DISABLED)
+ ErrDelegationTokenAuthDisabled ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_DELEGATION_TOKEN_AUTH_DISABLED)
// ErrDelegationTokenNotFound Broker: Delegation Token is not found on server
- ErrDelegationTokenNotFound ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_DELEGATION_TOKEN_NOT_FOUND)
+ ErrDelegationTokenNotFound ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_DELEGATION_TOKEN_NOT_FOUND)
// ErrDelegationTokenOwnerMismatch Broker: Specified Principal is not valid Owner/Renewer
- ErrDelegationTokenOwnerMismatch ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_DELEGATION_TOKEN_OWNER_MISMATCH)
+ ErrDelegationTokenOwnerMismatch ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_DELEGATION_TOKEN_OWNER_MISMATCH)
// ErrDelegationTokenRequestNotAllowed Broker: Delegation Token requests are not allowed on this connection
- ErrDelegationTokenRequestNotAllowed ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_DELEGATION_TOKEN_REQUEST_NOT_ALLOWED)
+ ErrDelegationTokenRequestNotAllowed ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_DELEGATION_TOKEN_REQUEST_NOT_ALLOWED)
// ErrDelegationTokenAuthorizationFailed Broker: Delegation Token authorization failed
- ErrDelegationTokenAuthorizationFailed ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_DELEGATION_TOKEN_AUTHORIZATION_FAILED)
+ ErrDelegationTokenAuthorizationFailed ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_DELEGATION_TOKEN_AUTHORIZATION_FAILED)
// ErrDelegationTokenExpired Broker: Delegation Token is expired
- ErrDelegationTokenExpired ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_DELEGATION_TOKEN_EXPIRED)
+ ErrDelegationTokenExpired ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_DELEGATION_TOKEN_EXPIRED)
// ErrInvalidPrincipalType Broker: Supplied principalType is not supported
- ErrInvalidPrincipalType ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INVALID_PRINCIPAL_TYPE)
+ ErrInvalidPrincipalType ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INVALID_PRINCIPAL_TYPE)
// ErrNonEmptyGroup Broker: The group is not empty
- ErrNonEmptyGroup ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_NON_EMPTY_GROUP)
+ ErrNonEmptyGroup ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_NON_EMPTY_GROUP)
// ErrGroupIDNotFound Broker: The group id does not exist
- ErrGroupIDNotFound ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_GROUP_ID_NOT_FOUND)
+ ErrGroupIDNotFound ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_GROUP_ID_NOT_FOUND)
// ErrFetchSessionIDNotFound Broker: The fetch session ID was not found
- ErrFetchSessionIDNotFound ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_FETCH_SESSION_ID_NOT_FOUND)
+ ErrFetchSessionIDNotFound ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_FETCH_SESSION_ID_NOT_FOUND)
// ErrInvalidFetchSessionEpoch Broker: The fetch session epoch is invalid
- ErrInvalidFetchSessionEpoch ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INVALID_FETCH_SESSION_EPOCH)
+ ErrInvalidFetchSessionEpoch ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INVALID_FETCH_SESSION_EPOCH)
// ErrListenerNotFound Broker: No matching listener
- ErrListenerNotFound ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_LISTENER_NOT_FOUND)
+ ErrListenerNotFound ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_LISTENER_NOT_FOUND)
// ErrTopicDeletionDisabled Broker: Topic deletion is disabled
- ErrTopicDeletionDisabled ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_TOPIC_DELETION_DISABLED)
+ ErrTopicDeletionDisabled ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_TOPIC_DELETION_DISABLED)
// ErrFencedLeaderEpoch Broker: Leader epoch is older than broker epoch
- ErrFencedLeaderEpoch ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_FENCED_LEADER_EPOCH)
+ ErrFencedLeaderEpoch ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_FENCED_LEADER_EPOCH)
// ErrUnknownLeaderEpoch Broker: Leader epoch is newer than broker epoch
- ErrUnknownLeaderEpoch ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_UNKNOWN_LEADER_EPOCH)
+ ErrUnknownLeaderEpoch ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_UNKNOWN_LEADER_EPOCH)
// ErrUnsupportedCompressionType Broker: Unsupported compression type
- ErrUnsupportedCompressionType ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_UNSUPPORTED_COMPRESSION_TYPE)
+ ErrUnsupportedCompressionType ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_UNSUPPORTED_COMPRESSION_TYPE)
// ErrStaleBrokerEpoch Broker: Broker epoch has changed
- ErrStaleBrokerEpoch ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_STALE_BROKER_EPOCH)
+ ErrStaleBrokerEpoch ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_STALE_BROKER_EPOCH)
// ErrOffsetNotAvailable Broker: Leader high watermark is not caught up
- ErrOffsetNotAvailable ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_OFFSET_NOT_AVAILABLE)
+ ErrOffsetNotAvailable ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_OFFSET_NOT_AVAILABLE)
// ErrMemberIDRequired Broker: Group member needs a valid member ID
- ErrMemberIDRequired ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_MEMBER_ID_REQUIRED)
+ ErrMemberIDRequired ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_MEMBER_ID_REQUIRED)
// ErrPreferredLeaderNotAvailable Broker: Preferred leader was not available
- ErrPreferredLeaderNotAvailable ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_PREFERRED_LEADER_NOT_AVAILABLE)
+ ErrPreferredLeaderNotAvailable ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_PREFERRED_LEADER_NOT_AVAILABLE)
// ErrGroupMaxSizeReached Broker: Consumer group has reached maximum size
- ErrGroupMaxSizeReached ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_GROUP_MAX_SIZE_REACHED)
+ ErrGroupMaxSizeReached ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_GROUP_MAX_SIZE_REACHED)
// ErrFencedInstanceID Broker: Static consumer fenced by other consumer with same group.instance.id
- ErrFencedInstanceID ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_FENCED_INSTANCE_ID)
+ ErrFencedInstanceID ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_FENCED_INSTANCE_ID)
// ErrEligibleLeadersNotAvailable Broker: Eligible partition leaders are not available
- ErrEligibleLeadersNotAvailable ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_ELIGIBLE_LEADERS_NOT_AVAILABLE)
+ ErrEligibleLeadersNotAvailable ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_ELIGIBLE_LEADERS_NOT_AVAILABLE)
// ErrElectionNotNeeded Broker: Leader election not needed for topic partition
- ErrElectionNotNeeded ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_ELECTION_NOT_NEEDED)
+ ErrElectionNotNeeded ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_ELECTION_NOT_NEEDED)
// ErrNoReassignmentInProgress Broker: No partition reassignment is in progress
- ErrNoReassignmentInProgress ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_NO_REASSIGNMENT_IN_PROGRESS)
+ ErrNoReassignmentInProgress ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_NO_REASSIGNMENT_IN_PROGRESS)
// ErrGroupSubscribedToTopic Broker: Deleting offsets of a topic while the consumer group is subscribed to it
- ErrGroupSubscribedToTopic ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_GROUP_SUBSCRIBED_TO_TOPIC)
+ ErrGroupSubscribedToTopic ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_GROUP_SUBSCRIBED_TO_TOPIC)
// ErrInvalidRecord Broker: Broker failed to validate record
- ErrInvalidRecord ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INVALID_RECORD)
+ ErrInvalidRecord ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INVALID_RECORD)
// ErrUnstableOffsetCommit Broker: There are unstable offsets that need to be cleared
- ErrUnstableOffsetCommit ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_UNSTABLE_OFFSET_COMMIT)
+ ErrUnstableOffsetCommit ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_UNSTABLE_OFFSET_COMMIT)
// ErrThrottlingQuotaExceeded Broker: Throttling quota has been exceeded
- ErrThrottlingQuotaExceeded ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_THROTTLING_QUOTA_EXCEEDED)
+ ErrThrottlingQuotaExceeded ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_THROTTLING_QUOTA_EXCEEDED)
// ErrProducerFenced Broker: There is a newer producer with the same transactionalId which fences the current one
- ErrProducerFenced ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_PRODUCER_FENCED)
+ ErrProducerFenced ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_PRODUCER_FENCED)
// ErrResourceNotFound Broker: Request illegally referred to resource that does not exist
- ErrResourceNotFound ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_RESOURCE_NOT_FOUND)
+ ErrResourceNotFound ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_RESOURCE_NOT_FOUND)
// ErrDuplicateResource Broker: Request illegally referred to the same resource twice
- ErrDuplicateResource ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_DUPLICATE_RESOURCE)
+ ErrDuplicateResource ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_DUPLICATE_RESOURCE)
// ErrUnacceptableCredential Broker: Requested credential would not meet criteria for acceptability
- ErrUnacceptableCredential ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_UNACCEPTABLE_CREDENTIAL)
+ ErrUnacceptableCredential ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_UNACCEPTABLE_CREDENTIAL)
// ErrInconsistentVoterSet Broker: Indicates that the either the sender or recipient of a voter-only request is not one of the expected voters
- ErrInconsistentVoterSet ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INCONSISTENT_VOTER_SET)
+ ErrInconsistentVoterSet ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INCONSISTENT_VOTER_SET)
// ErrInvalidUpdateVersion Broker: Invalid update version
- ErrInvalidUpdateVersion ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INVALID_UPDATE_VERSION)
+ ErrInvalidUpdateVersion ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_INVALID_UPDATE_VERSION)
// ErrFeatureUpdateFailed Broker: Unable to update finalized features due to server error
- ErrFeatureUpdateFailed ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_FEATURE_UPDATE_FAILED)
+ ErrFeatureUpdateFailed ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_FEATURE_UPDATE_FAILED)
// ErrPrincipalDeserializationFailure Broker: Request principal deserialization failed during forwarding
- ErrPrincipalDeserializationFailure ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_PRINCIPAL_DESERIALIZATION_FAILURE)
+ ErrPrincipalDeserializationFailure ErrorCode = ErrorCode(C.RD_KAFKA_RESP_ERR_PRINCIPAL_DESERIALIZATION_FAILURE)
)
func (c ErrorCode) String() string+
func (c ErrorCode) String() string
String returns a human readable representation of an error code
type Event interface {
// String returns a human-readable representation of the event
- String() string
+ String() string
}
type Header struct {
- Key string // Header name (utf-8 string)
- Value []byte // Header value (nil, empty, or binary)
+ Key string // Header name (utf-8 string)
+ Value []byte // Header value (nil, empty, or binary)
}
func (h Header) String() string+
func (h Header) String() string
String returns the Header Key and data in a human representable possibly truncated form suitable for displaying to the user.
type LogEvent struct {
- Name string // Name of client instance
- Tag string // Log tag that provides context to the log Message (e.g., "METADATA" or "GRPCOORD")
- Message string // Log message
- Level int // Log syslog level, lower is more critical.
- Timestamp time.Time // Log timestamp
+ Name string // Name of client instance
+ Tag string // Log tag that provides context to the log Message (e.g., "METADATA" or "GRPCOORD")
+ Message string // Log message
+ Level int // Log syslog level, lower is more critical.
+ Timestamp time.Time // Log timestamp
}
func (logEvent LogEvent) String() string+
func (logEvent LogEvent) String() string
type Message struct {
TopicPartition TopicPartition
- Value []byte
- Key []byte
- Timestamp time.Time
+ Value []byte
+ Key []byte
+ Timestamp time.Time
TimestampType TimestampType
Opaque interface{}
Headers []Header
@@ -4166,21 +4166,21 @@
func (m *Message) String() string+
func (m *Message) String() string
String returns a human readable representation of a Message. Key and payload are not represented.
type Metadata struct {
Brokers []BrokerMetadata
- Topics map[string]TopicMetadata
+ Topics map[string]TopicMetadata
OriginatingBroker BrokerMetadata
}
func NewMockCluster(brokerCount int) (*MockCluster, error)+
func NewMockCluster(brokerCount int) (*MockCluster, error)
NewMockCluster provides a mock Kafka cluster with a configurable number of brokers that support a reasonable subset of Kafka protocol @@ -4247,20 +4247,20 @@
func (mc *MockCluster) BootstrapServers() string+
func (mc *MockCluster) BootstrapServers() string
BootstrapServers returns the bootstrap.servers property for this MockCluster
type OAuthBearerTokenRefresh struct {
// Config is the value of the sasl.oauthbearer.config property
- Config string
+ Config string
}
func (o OAuthBearerTokenRefresh) String() string+
func (o OAuthBearerTokenRefresh) String() string
Offset type (int64) with support for canonical names
-type Offset int64+
type Offset int64
func NewOffset(offset interface{}) (Offset, error)
+ func NewOffset(offset interface{}) (Offset, error)
NewOffset creates a new Offset using the provided logical string, or an absolute int64 offset value. @@ -4360,7 +4360,7 @@
func (o *Offset) Set(offset interface{}) error+
func (o *Offset) Set(offset interface{}) error
Set offset value, see NewOffset()
func (o Offset) String() string+
func (o Offset) String() string
type OffsetsCommitted struct {
- Error error
+ Error error
Offsets []TopicPartition
}
func (o OffsetsCommitted) String() string+
func (o OffsetsCommitted) String() string
type PartitionEOF TopicPartition
func (p PartitionEOF) String() string+
func (p PartitionEOF) String() string
type PartitionMetadata struct {
- ID int32
+ ID int32
Error Error
- Leader int32
- Replicas []int32
- Isrs []int32
+ Leader int32
+ Replicas []int32
+ Isrs []int32
}
type PartitionsSpecification struct {
// Topic to create more partitions for.
- Topic string
+ Topic string
// New partition count for topic, must be higher than current partition count.
- IncreaseTo int
+ IncreaseTo int
// (Optional) Explicit replica assignment. The outer array is
// indexed by the new partition index (i.e., 0 for the first added
// partition), while the inner per-partition array
// contains the replica broker ids. The first broker in each
// broker id list will be the preferred replica.
- ReplicaAssignment [][]int32
+ ReplicaAssignment [][]int32
}
func NewProducer(conf *ConfigMap) (*Producer, error)+
func NewProducer(conf *ConfigMap) (*Producer, error)
NewProducer creates a new high-level Producer instance.
@@ -4542,14 +4542,14 @@func (p *Producer) AbortTransaction(ctx context.Context) error+
func (p *Producer) AbortTransaction(ctx context.Context) error
AbortTransaction aborts the ongoing transaction.
@@ -4587,14 +4587,14 @@func (p *Producer) BeginTransaction() error+
func (p *Producer) BeginTransaction() error
BeginTransaction starts a new transaction.
@@ -4624,7 +4624,7 @@func (p *Producer) CommitTransaction(ctx context.Context) error+
func (p *Producer) CommitTransaction(ctx context.Context) error
CommitTransaction commits the current transaction.
@@ -4688,7 +4688,7 @@func (p *Producer) Flush(timeoutMs int) int+
func (p *Producer) Flush(timeoutMs int) int
Flush and wait for outstanding messages and requests to complete delivery. Includes messages on ProduceChannel. @@ -4717,27 +4717,27 @@
func (p *Producer) GetFatalError() error+
func (p *Producer) GetFatalError() error
GetFatalError returns an Error object if the client instance has raised a fatal error, else nil.
func (p *Producer) GetMetadata(topic *string, allTopics bool, timeoutMs int) (*Metadata, error)+
func (p *Producer) GetMetadata(topic *string, allTopics bool, timeoutMs int) (*Metadata, error)
GetMetadata queries broker for cluster and topic metadata. If topic is non-nil only information about that topic is returned, else if @@ -4747,14 +4747,14 @@
func (p *Producer) InitTransactions(ctx context.Context) error+
func (p *Producer) InitTransactions(ctx context.Context) error
InitTransactions Initializes transactions for the producer instance.
@@ -4802,14 +4802,14 @@func (p *Producer) Len() int+
func (p *Producer) Len() int
Len returns the number of messages and requests waiting to be transmitted to the broker as well as delivery reports queued for the application. @@ -4817,7 +4817,7 @@
func (p *Producer) OffsetsForTimes(times []TopicPartition, timeoutMs int) (offsets []TopicPartition, err error)+
func (p *Producer) OffsetsForTimes(times []TopicPartition, timeoutMs int) (offsets []TopicPartition, err error)
OffsetsForTimes looks up offsets by timestamp for the given partitions.
@@ -4861,14 +4861,14 @@func (p *Producer) Produce(msg *Message, deliveryChan chan Event) error+
func (p *Producer) Produce(msg *Message, deliveryChan chan Event) error
Produce single message. This is an asynchronous call that enqueues the message on the internal @@ -4883,7 +4883,7 @@
func (p *Producer) Purge(flags int) error+
func (p *Producer) Purge(flags int) error
Purge messages currently handled by this producer instance.
@@ -4937,28 +4937,28 @@func (p *Producer) QueryWatermarkOffsets(topic string, partition int32, timeoutMs int) (low, high int64, err error)+
func (p *Producer) QueryWatermarkOffsets(topic string, partition int32, timeoutMs int) (low, high int64, err error)
QueryWatermarkOffsets returns the broker's low and high offsets for the given topic and partition.
func (p *Producer) SendOffsetsToTransaction(ctx context.Context, offsets []TopicPartition, consumerMetadata *ConsumerGroupMetadata) error+
func (p *Producer) SendOffsetsToTransaction(ctx context.Context, offsets []TopicPartition, consumerMetadata *ConsumerGroupMetadata) error
SendOffsetsToTransaction sends a list of topic partition offsets to the consumer group coordinator for `consumerMetadata`, and marks the offsets @@ -5006,14 +5006,14 @@
func (p *Producer) SetOAuthBearerToken(oauthBearerToken OAuthBearerToken) error+
func (p *Producer) SetOAuthBearerToken(oauthBearerToken OAuthBearerToken) error
SetOAuthBearerToken sets the the data to be transmitted to a broker during SASL/OAUTHBEARER authentication. It will return nil @@ -5031,14 +5031,14 @@
func (p *Producer) SetOAuthBearerTokenFailure(errstr string) error+
func (p *Producer) SetOAuthBearerTokenFailure(errstr string) error
SetOAuthBearerTokenFailure sets the error message describing why token retrieval/setting failed; it also schedules a new token refresh event for 10 @@ -5050,34 +5050,34 @@
func (p *Producer) String() string+
func (p *Producer) String() string
String returns a human readable name for a Producer instance
func (p *Producer) TestFatalError(code ErrorCode, str string) ErrorCode+
func (p *Producer) TestFatalError(code ErrorCode, str string) ErrorCode
TestFatalError triggers a fatal error in the underlying client. This is to be used strictly for testing purposes.
type RebalanceCb func(*Consumer, Event) error+
type RebalanceCb func(*Consumer, Event) error
ResourcePatternType enumerates the different types of Kafka resource patterns.
-type ResourcePatternType int+
type ResourcePatternType int
func ResourcePatternTypeFromString(patternTypeString string) (ResourcePatternType, error)+
func ResourcePatternTypeFromString(patternTypeString string) (ResourcePatternType, error)
ResourcePatternTypeFromString translates a resource pattern type name to a ResourcePatternType value.
func (t ResourcePatternType) String() string+
func (t ResourcePatternType) String() string
String returns the human-readable representation of a ResourcePatternType
ResourceType represents an Apache Kafka resource type
-type ResourceType int+
type ResourceType int
func ResourceTypeFromString(typeString string) (ResourceType, error)+
func ResourceTypeFromString(typeString string) (ResourceType, error)
ResourceTypeFromString translates a resource type name/string to a ResourceType value.
func (t ResourceType) String() string+
func (t ResourceType) String() string
String returns the human-readable representation of a ResourceType
func (e RevokedPartitions) String() string+
func (e RevokedPartitions) String() string
func (e Stats) String() string+
func (e Stats) String() string
TimestampType is a the Message timestamp type or source
-type TimestampType int+
type TimestampType int
func (t TimestampType) String() string+
func (t TimestampType) String() string
type TopicMetadata struct {
- Topic string
+ Topic string
Partitions []PartitionMetadata
Error Error
}
type TopicPartition struct {
- Topic *string
- Partition int32
+ Topic *string
+ Partition int32
Offset Offset
- Metadata *string
- Error error
+ Metadata *string
+ Error error
}
func (p TopicPartition) String() string+
func (p TopicPartition) String() string
type TopicPartitions []TopicPartition
func (tps TopicPartitions) Len() int+
func (tps TopicPartitions) Len() int
func (tps TopicPartitions) Less(i, j int) bool+
func (tps TopicPartitions) Less(i, j int) bool
func (tps TopicPartitions) Swap(i, j int)+
func (tps TopicPartitions) Swap(i, j int)
type TopicResult struct {
// Topic name
- Topic string
+ Topic string
// Error, if any, of result. Check with `Error.Code() != ErrNoError`.
Error Error
}
func (t TopicResult) String() string+
func (t TopicResult) String() string
String returns a human-readable representation of a TopicResult.
type TopicSpecification struct {
// Topic name to create.
- Topic string
+ Topic string
// Number of partitions in topic.
- NumPartitions int
+ NumPartitions int
// Default replication factor for the topic's partitions, or zero
// if an explicit ReplicaAssignment is set.
- ReplicationFactor int
+ ReplicationFactor int
// (Optional) Explicit replica assignment. The outer array is
// indexed by the partition number, while the inner per-partition array
// contains the replica broker ids. The first broker in each
// broker id list will be the preferred replica.
- ReplicaAssignment [][]int32
+ ReplicaAssignment [][]int32
// Topic configuration.
- Config map[string]string
+ Config map[string]string
}