kgo

func AllISRAcks ¶

func AllISRAcks() Acks

AllISRAcks ensures that all in-sync replicas have acknowledged they wrote a record before the leader replies success.

func LeaderAck ¶

func LeaderAck() Acks

LeaderAck causes Kafka to reply that a record is written after only the leader has written a message. The leader does not wait for in-sync replica replies.

func NoAck ¶

func NoAck() Acks

NoAck considers records sent as soon as they are written on the wire. The leader does not reply to records.

func NewClient ¶

func NewClient(opts ...Opt) (*Client, error)

NewClient returns a new Kafka client with the given options or an error if the options are invalid. Connections to brokers are lazily created only when requests are written to them.

By default, the client uses the latest stable request versions when talking to Kafka. If you use a broker older than 0.10.0, then you need to manually set a MaxVersions option. Otherwise, there is usually no harm in defaulting to the latest API versions, although occasionally Kafka introduces new required parameters that do not have zero value defaults.

NewClient also launches a goroutine which periodically updates the cached topic metadata.

func GzipCompression ¶

func GzipCompression() CompressionCodec

GzipCompression enables gzip compression with the default compression level.

func Lz4Compression ¶

func Lz4Compression() CompressionCodec

Lz4Compression enables lz4 compression with the fastest compression level.

func NoCompression ¶

func NoCompression() CompressionCodec

NoCompression is the default compression used for messages and can be used as a fallback compression option.

func SnappyCompression ¶

func SnappyCompression() CompressionCodec

SnappyCompression enables snappy compression.

func ZstdCompression ¶

func ZstdCompression() CompressionCodec

ZstdCompression enables zstd compression with the default compression level.

func NewConsumerBalancer ¶

func NewConsumerBalancer(balance ConsumerBalancerBalance, members []kmsg.JoinGroupResponseMember) (*ConsumerBalancer, error)

NewConsumerBalancer parses the each member's metadata as a kmsg.GroupMemberMetadata and returns a ConsumerBalancer to use in balancing.

If any metadata parsing fails, this returns an error.

func AllowedConcurrentFetches ¶

func AllowedConcurrentFetches(n int) ConsumerOpt

AllowedConcurrentFetches sets the maximum number of fetch requests to allow in flight or buffered at once, overriding the unbounded (i.e. number of brokers) default.

This setting, paired with FetchMaxBytes, can upper bound the maximum amount of memory that the client can use for consuming.

Requests are issued to brokers in a FIFO order: once the client is ready to issue a request to a broker, it registers that request and issues it in order with other registrations.

If Kafka replies with any data, the client does not track the fetch as completed until the user has polled the buffered fetch. Thus, a concurrent fetch is not considered complete until all data from it is done being processed and out of the client itself.

Note that brokers are allowed to hang for up to FetchMaxWait before replying to a request, so if this option is too constrained and you are consuming a low throughput topic, the client may take a long time before requesting a broker that has new data. For high throughput topics, or if the allowed concurrent fetches is large enough, this should not be a concern.

A value of 0 implies the allowed concurrency is unbounded and will be limited only by the number of brokers in the cluster.

func ConsumeResetOffset ¶

func ConsumeResetOffset(offset Offset) ConsumerOpt

ConsumeResetOffset sets the offset to restart consuming from when a partition has no commits (for groups) or when a fetch sees an OffsetOutOfRange error, overriding the default ConsumeStartOffset.

func FetchIsolationLevel ¶

func FetchIsolationLevel(level IsolationLevel) ConsumerOpt

FetchIsolationLevel sets the "isolation level" used for fetching records, overriding the default ReadUncommitted.

func FetchMaxBytes ¶

func FetchMaxBytes(b int32) ConsumerOpt

FetchMaxBytes sets the maximum amount of bytes a broker will try to send during a fetch, overriding the default 50MiB. Note that brokers may not obey this limit if it has records larger than this limit. Also note that this client sends a fetch to each broker concurrently, meaning the client will buffer up to <brokers * max bytes> worth of memory.

This corresponds to the Java fetch.max.bytes setting.

If bumping this, consider bumping BrokerMaxReadBytes.

func FetchMaxPartitionBytes ¶

func FetchMaxPartitionBytes(b int32) ConsumerOpt

FetchMaxPartitionBytes sets the maximum amount of bytes that will be consumed for a single partition in a fetch request, overriding the default 10MiB. Note that if a single batch is larger than this number, that batch will still be returned so the client can make progress.

This corresponds to the Java max.partition.fetch.bytes setting.

func FetchMaxWait ¶

func FetchMaxWait(wait time.Duration) ConsumerOpt

FetchMaxWait sets the maximum amount of time a broker will wait for a fetch response to hit the minimum number of required bytes before returning, overriding the default 5s.

This corresponds to the Java replica.fetch.wait.max.ms setting.

func FetchMinBytes ¶

func FetchMinBytes(b int32) ConsumerOpt

FetchMinBYtes sets the minimum amount of bytes a broker will try to send during a fetch, overriding the default 1 byte.

With the default of 1, data is sent as soon as it is available. By bumping this, the broker will try to wait for more data, which may improve server throughput at the expense of added latency.

This corresponds to the Java fetch.min.bytes setting.

func KeepControlRecords ¶

func KeepControlRecords() ConsumerOpt

KeepControlRecords sets the client to keep control messages and return them with fetches, overriding the default that discards them.

Generally, control messages are not useful.

func Rack ¶

func Rack(rack string) ConsumerOpt

Rack specifies where the client is physically located and changes fetch requests to consume from the closest replica as opposed to the leader replica.

Consuming from a preferred replica can increase latency but can decrease cross datacenter costs. See KIP-392 for more information.

func ConsumePartitions ¶

func ConsumePartitions(partitions map[string]map[int32]Offset) DirectConsumeOpt

ConsumePartitions sets partitions to consume from directly and the offsets to start consuming those partitions from.

Offsets from option have higher precedence than ConsumeTopics. If a topic's partition is set in this option and that topic is also set in ConsumeTopics, offsets on partitions in this option are used in favor of the more general topic offset from ConsumeTopics.

func ConsumeTopics ¶

func ConsumeTopics(offset Offset, topics ...string) DirectConsumeOpt

ConsumeTopics sets topics to consume directly and the offsets to start consuming partitions from in those topics.

If a metadata update sees partitions added to a topic, the client will automatically begin consuming from those new partitions.

func ConsumeTopicsRegex ¶

func ConsumeTopicsRegex() DirectConsumeOpt

ConsumeTopicsRegex sets all topics in ConsumeTopics to be parsed as regular expressions.

func AbortingFirstErrPromise ¶

func AbortingFirstErrPromise(cl *Client) *FirstErrPromise

AbortingFirstErrPromise returns a FirstErrPromise that will call the client's AbortBufferedRecords function if an error is encountered.

This can be used to quickly exit when any error is encountered, rather than waiting while flushing only to discover things errored.

func CooperativeStickyBalancer ¶

func CooperativeStickyBalancer() GroupBalancer

CooperativeStickyBalancer performs the sticky balancing strategy, but additionally opts the consumer group into "cooperative" rebalancing.

Cooperative rebalancing differs from "eager" (the original) rebalancing in that group members do not stop processing partitions during the rebalance. Instead, once they receive their new assignment, each member determines which partitions it needs to revoke. If any, they send a new join request (before syncing), and the process starts over. This should ultimately end up in only two join rounds, with the major benefit being that processing never needs to stop.

NOTE once a group is collectively using cooperative balancing, it is unsafe to have a member join the group that does not support cooperative balancing. If the only-eager member is elected leader, it will not know of the new multiple join strategy and things will go awry. Thus, once a group is entirely on cooperative rebalancing, it cannot go back.

Migrating an eager group to cooperative balancing requires two rolling bounce deploys. The first deploy should add the cooperative-sticky strategy as an option (that is, each member goes from using one balance strategy to two). During this deploy, Kafka will tell leaders to continue using the old eager strategy, since the old eager strategy is the only one in common among all members. The second rolling deploy removes the old eager strategy. At this point, Kafka will tell the leader to use cooperative-sticky balancing. During this roll, all members in the group that still have both strategies continue to be eager and give up all of their partitions every rebalance. However, once a member only has cooperative-sticky, it can begin using this new strategy and things will work correctly. See KIP-429 for more details.

func RangeBalancer ¶

func RangeBalancer() GroupBalancer

RangeBalancer returns a group balancer that, per topic, maps partitions to group members. Since this works on a topic level, uneven partitions per topic to the number of members can lead to slight partition consumption disparities.

Suppose there are two members M0 and M1, two topics t0 and t1, and each topic has three partitions p0, p1, and p2. The partition balancing will be

M0: [t0p0, t0p1, t1p0, t1p1]
M1: [t0p2, t1p2]

This is equivalent to the Java range balancer.

func RoundRobinBalancer ¶

func RoundRobinBalancer() GroupBalancer

RoundRobinBalancer returns a group balancer that evenly maps topics and partitions to group members.

Suppose there are two members M0 and M1, two topics t0 and t1, and each topic has three partitions p0, p1, and p2. The partition balancing will be

M0: [t0p0, t0p2, t1p1]
M1: [t0p1, t1p0, t1p2]

If all members subscribe to all topics equally, the roundrobin balancer will give a perfect balance. However, if topic subscriptions are quite unequal, the roundrobin balancer may lead to a bad balance. See KIP-49 for one example (note that the fair strategy mentioned in KIP-49 does not exist).

This is equivalent to the Java roundrobin balancer.

func StickyBalancer ¶

func StickyBalancer() GroupBalancer

StickyBalancer returns a group balancer that ensures minimal partition movement on group changes while also ensuring optimal balancing.

Suppose there are three members M0, M1, and M3, and two topics t0 and t1 each with three partitions p0, p1, and p2. If the initial balance plan looks like

M0: [t0p0, t0p1, t0p2]
M1: [t1p0, t1p1, t1p2]
M2: [t2p0, t2p2, t2p2]

If M2 disappears, both roundrobin and range would have mostly destructive reassignments.

Range would result in

M0: [t0p0, t0p1, t1p0, t1p1, t2p0, t2p1]
M1: [t0p2, t1p2, t2p2]

which is imbalanced and has 3 partitions move from members that did not need to move (t0p2, t1p0, t1p1).

RoundRobin would result in

M0: [t0p0, t0p2, t1p1, t2p0, t2p2]
M1: [t0p1, t1p0, t1p2, t2p1]

which is balanced, but has 2 partitions move when they do not need to (t0p1, t1p1).

Sticky balancing results in

M0: [t0p0, t0p1, t0p2, t2p0, t2p2]
M1: [t1p0, t1p1, t1p2, t2p1]

which is balanced and does not cause any unnecessary partition movement. The actual t2 partitions may not be in that exact combination, but they will be balanced.

An advantage of the sticky consumer is that it allows API users to potentially avoid some cleanup until after the consumer knows which partitions it is losing when it gets its new assignment. Users can then only cleanup state for partitions that changed, which will be minimal (see KIP-54; this client also includes the KIP-351 bugfix).

Note that this API implements the sticky partitioning quite differently from the Java implementation. The Java implementaiton is difficult to reason about and has many edge cases that result in non-optimal balancing (albeit, you likely have to be trying to hit those edge cases). This API uses a different algorithm to ensure optimal balancing while being an order of magnitude faster.

Since the new strategy is a strict improvement over the Java strategy, it is entirely compatible. Any Go client sharing a group with a Java client will not have its decisions undone on leadership change from a Go consumer to a Java one. Java balancers do not apply the strategy it comes up with if it deems the balance score equal to or worse than the original score (the score being effectively equal to the standard deviation of the mean number of assigned partitions). This Go sticky balancer is optimal and extra sticky. Thus, the Java balancer will never back out of a strategy from this balancer.

func AutoCommitInterval ¶

func AutoCommitInterval(interval time.Duration) GroupOpt

AutoCommitInterval sets how long to go between autocommits, overriding the default 5s.

func Balancers ¶

func Balancers(balancers ...GroupBalancer) GroupOpt

Balancers sets the group balancers to use for dividing topic partitions among group members, overriding the defaults.

The current default is [cooperative-sticky].

For balancing, Kafka chooses the first protocol that all group members agree to support.

Note that if you want to opt in to cooperative-sticky rebalancing, cooperative group balancing is incompatible with eager (classical) rebalancing and requires a careful rollout strategy (see KIP-429).

func CommitCallback ¶

func CommitCallback(fn func(*kmsg.OffsetCommitRequest, *kmsg.OffsetCommitResponse, error)) GroupOpt

CommitCallback sets the callback to use if autocommitting is enabled. This overrides the default callback that logs errors and continues.

func DisableAutoCommit ¶

func DisableAutoCommit() GroupOpt

DisableAutoCommit disable auto committing.

func GroupProtocol ¶

func GroupProtocol(protocol string) GroupOpt

GroupProtocol sets the group's join protocol, overriding the default value "consumer". The only reason to override this is if you are implementing custom join and sync group logic.

func GroupTopics ¶

func GroupTopics(topics ...string) GroupOpt

GroupTopics adds topics to use for group consuming.

func GroupTopicsRegex ¶

func GroupTopicsRegex() GroupOpt

GroupTopicsRegex sets all topics in GroupTopics to be parsed as regular expressions.

func HeartbeatInterval ¶

func HeartbeatInterval(interval time.Duration) GroupOpt

HeartbeatInterval sets how long a group member goes between heartbeats to Kafka, overriding the default 3,000ms.

Kafka uses heartbeats to ensure that a group member's session stays active. This value can be any value lower than the session timeout, but should be no higher than 1/3rd the session timeout.

This corresponds to Kafka's heartbeat.interval.ms.

func InstanceID ¶

func InstanceID(id string) GroupOpt

InstanceID sets the group consumer's instance ID, switching the group member from "dynamic" to "static".

Prior to Kafka 2.3.0, joining a group gave a group member a new member ID. The group leader could not tell if this was a rejoining member. Thus, any join caused the group to rebalance.

Kafka 2.3.0 introduced the concept of an instance ID, which can persist across restarts. This allows for avoiding many costly rebalances and allows for stickier rebalancing for rejoining members (since the ID for balancing stays the same). The main downsides are that you, the user of a client, have to manage instance IDs properly, and that it may take longer to rebalance in the event that a client legitimately dies.

When using an instance ID, the client does NOT send a leave group request when closing. This allows for the client ot restart with the same instance ID and rejoin the group to avoid a rebalance. It is strongly recommended to increase the session timeout enough to allow time for the restart (remember that the default session timeout is 10s).

To actually leave the group, you must use an external admin command that issues a leave group request on behalf of this instance ID (see kcl), or you can manually use the kmsg package with a proper LeaveGroupRequest.

NOTE: Leaving a group with an instance ID is only supported in Kafka 2.4.0+.

func OnAssigned ¶

func OnAssigned(onAssigned func(context.Context, map[string][]int32)) GroupOpt

OnAssigned sets the function to be called when a group is joined after partitions are assigned before fetches for those partitions begin.

This function combined with OnRevoked should not exceed the rebalance interval. It is possible for the group, immediately after finishing a balance, to re-enter a new balancing session.

The OnAssigned function is passed the group's context, which is only canceled if the group is left or the client is closed.

func OnLost ¶

func OnLost(onLost func(context.Context, map[string][]int32)) GroupOpt

OnLost sets the function to be called on "fatal" group errors, such as IllegalGeneration, UnknownMemberID, and authentication failures. This function differs from OnRevoked in that it is unlikely that commits will succeed when partitions are outright lost, whereas commits likely will succeed when revoking partitions.

If not set, OnRevoked is used.

func OnRevoked ¶

func OnRevoked(onRevoked func(context.Context, map[string][]int32)) GroupOpt

OnRevoked sets the function to be called once this group member has partitions revoked.

This function combined with OnAssigned should not exceed the rebalance interval. It is possible for the group, immediately after finishing a balance, to re-enter a new balancing session.

If autocommit is enabled, the default OnRevoked is a blocking commit all offsets. The reason for a blocking commit is so that no later commit cancels the blocking commit. If the commit in OnRevoked were canceled, then the rebalance would proceed immediately, the commit that canceled the blocking commit would fail, and duplicates could be consumed after the rebalance completes.

The OnRevoked function is passed the group's context, which is only canceled if the group is left or the client is closed.

OnRevoked function is called at the end of a group session even if there are no partitions being revoked.

If you are committing offsets manually (have disabled autocommitting), it is highly recommended to do a proper blocking commit in OnRevoked.

func RebalanceTimeout ¶

func RebalanceTimeout(timeout time.Duration) GroupOpt

RebalanceTimeout sets how long group members are allowed to take when a a rebalance has begun, overriding the default 60,000ms. This timeout is how long all members are allowed to complete work and commit offsets, minus the time it took to detect the rebalance (from a heartbeat).

Kafka uses the largest rebalance timeout of all members in the group. If a member does not rejoin within this timeout, Kafka will kick that member from the group.

This corresponds to Kafka's rebalance.timeout.ms.

func RequireStableFetchOffsets ¶

func RequireStableFetchOffsets() GroupOpt

RequireStableFetchOffsets sets the group consumer to require "stable" fetch offsets before consuming from the group. Proposed in KIP-447 and introduced in Kafka 2.5.0, stable offsets are important when consuming from partitions that a transactional producer could be committing to.

With this option, Kafka will block group consumers from fetching offsets for partitions that are in an active transaction.

Because this can block consumption, it is strongly recommended to set transactional timeouts to a small value (10s) rather than the default 60s. Lowering the transactional timeout will reduce the chance that consumers are entirely blocked.

func SessionTimeout ¶

func SessionTimeout(timeout time.Duration) GroupOpt

SessionTimeout sets how long a member the group can go between heartbeats, overriding the default 45,000ms. If a member does not heartbeat in this timeout, the broker will remove the member from the group and initiate a rebalance.

If you are using a GroupTransactSession for EOS, wish to lower this, and are talking to a Kafka cluster pre 2.5.0, consider lowering the TransactionTimeout. If you do not, you risk a transaction finishing after a group has rebalanced, which could lead to duplicate processing. If you are talking to a Kafka 2.5.0+ cluster, you can safely use the RequireStableFetchOffsets group option and prevent any problems.

This option corresponds to Kafka's session.timeout.ms setting and must be within the broker's group.min.session.timeout.ms and group.max.session.timeout.ms.

func ReadCommitted ¶

func ReadCommitted() IsolationLevel

ReadCommitted is an isolation level to only fetch committed records.

func ReadUncommitted ¶

func ReadUncommitted() IsolationLevel

ReadUncommitted (the default) is an isolation level that returns the latest produced records, be they committed or not.

func BasicLogger ¶

func BasicLogger(dst io.Writer, level LogLevel, prefixFn func() string) Logger

BasicLogger returns a logger that will print to dst in the following format:

prefix [LEVEL] message; key: val, key: val

prefixFn is optional; if non-nil, it is called for a per-message prefix.

Writes to dst are not checked for errors.

func NewOffset ¶

func NewOffset() Offset

NewOffsetcreates and returns an offset to use in AssignPartitions.

The default offset begins at the end.

func AutoTopicCreation ¶

func AutoTopicCreation() Opt

AutoTopicCreation enables topics to be auto created if they do not exist when fetching their metadata.

func BrokerConnDeadRetries ¶

func BrokerConnDeadRetries(n int) Opt

BrokerConnDeadRetries sets the number of tries that are allowed for requests that fail before a response is even received, overriding the default 20.

These retries are for cases where a connection to a broker fails during a request. Generally, you just must retry these requests anyway. However, if Kafka cannot understand our request, it closes a connection deliberately. Since the client cannot differentiate this case, we upper bound the number of times we allow a connection to be cut before we call it quits on the request, assuming that if a connection is cut *so many times* repeatedly, then it is likely not the network but instead Kafka indicating a problem.

This can only be triggered by connection read or write failures, indicating Kafka closed the connection (or the connection actually just died).

This setting applies to all but internally generated fetch and produce requests.

func BrokerMaxReadBytes ¶

func BrokerMaxReadBytes(v int32) Opt

BrokerMaxReadBytes sets the maximum response size that can be read from Kafka, overriding the default 100MiB.

This is a safety measure to avoid OOMing on invalid responses. This is slightly double FetchMaxBytes; if bumping that, consider bump this. No other response should run the risk of hitting this limit.

func BrokerMaxWriteBytes ¶

func BrokerMaxWriteBytes(v int32) Opt

BrokerMaxWriteBytes upper bounds the number of bytes written to a broker connection in a single write, overriding the default 100MiB.

This number corresponds to the a broker's socket.request.max.bytes, which defaults to 100MiB.

The only Kafka request that could come reasonable close to hitting this limit should be produce requests, and thus this limit is only enforced for produce requests.

func ClientID ¶

func ClientID(id string) Opt

ClientID uses id for all requests sent to Kafka brokers, overriding the default "kgo".

func ConnIdleTimeout ¶

func ConnIdleTimeout(timeout time.Duration) Opt

ConnIdleTimeout is a rough amount of time to allow connections to idle before they are closed, overriding the default 20.

In the worst case, a connection can be allowed to idle for up to 2x this time, while the average is expected to be 1.5x (essentially, a uniform distribution from this interval to 2x the interval).

It is possible that a connection can be reaped just as it is about to be written to, but the client internally retries in these cases.

Connections are not reaped if they are actively being written to or read from; thus, a request can take a really long time itself and not be reaped (however, this may lead to the ConnTimeoutOverhead).

func ConnTimeoutOverhead ¶

func ConnTimeoutOverhead(overhead time.Duration) Opt

ConnTimeoutOverhead uses the given time as overhead while deadlining requests, overriding the default overhead of 20s.

For most requests, the overhead will simply be this timeout. However, for any request with a TimeoutMillis field, the overhead is added on top of the request's TimeoutMillis. This ensures that we give Kafka enough time to actually process the request given the timeout, while still having a deadline on the connection as a whole to ensure it does not hang.

For writes, the timeout is always the overhead. We buffer writes in our client before one quick flush, so we always expect the write to be fast.

func Dialer ¶

func Dialer(fn func(ctx context.Context, network, host string) (net.Conn, error)) Opt

Dialer uses fn to dial addresses, overriding the default dialer that uses a 10s dial timeout and no TLS.

The context passed to the dial function is the context used in the request that caused the dial. If the request is a client-internal request, the context is the context on the client itself (which is canceled when the client is closed).

This function has the same signature as net.Dialer's DialContext and tls.Dialer's DialContext, meaning you can use this function like so:

kgo.Dialer((&net.Dialer{Timeout: 10*time.Second}).DialContext)

or

kgo.Dialer((&tls.Dialer{...})}.DialContext)

func DisableClientID ¶

func DisableClientID() Opt

DisableClientID sets the client ID to null for all requests sent to Kafka brokers, overriding the default "kgo".

func MaxVersions ¶

func MaxVersions(versions *kversion.Versions) Opt

MaxVersions sets the maximum Kafka version to try, overriding the internal unbounded (latest stable) versions.

Note that specific max version pinning is required if trying to interact with versions pre 0.10.0. Otherwise, unless using more complicated requests that this client itself does not natively use, it is generally safe to opt for the latest version. If using the kmsg package directly to issue requests, it is recommended to pin versions so that new fields on requests do not get invalid default zero values before you update your usage.

func MetadataMaxAge ¶

func MetadataMaxAge(age time.Duration) Opt

MetadataMaxAge sets the maximum age for the client's cached metadata, overriding the default 5m, to allow detection of new topics, partitions, etc.

This corresponds to Kafka's metadata.max.age.ms.

func MetadataMinAge ¶

func MetadataMinAge(age time.Duration) Opt

MetadataMinAge sets the minimum time between metadata queries, overriding the default 10s. You may want to raise or lower this to reduce the number of metadata queries the client will make. Notably, if metadata detects an error in any topic or partition, it triggers itself to update as soon as allowed. Additionally, any connection failures causing backoff while producing or consuming trigger metadata updates, because the client must assume that maybe the connection died due to a broker dying.

func MinVersions ¶

func MinVersions(versions *kversion.Versions) Opt

MinVersions sets the minimum Kafka version a request can be downgraded to, overriding the default of the lowest version.

This option is useful if you are issuing requests that you absolutely do not want to be downgraded; that is, if you are relying on features in newer requests, and you are not sure if your brokers can handle those features. By setting a min version, if the client detects it needs to downgrade past the version, it will instead avoid issuing the request.

Unlike MaxVersions, if a request is issued that is unknown to the min versions, the request is allowed. It is assumed that there is no lower bound for that request.

func ProduceRetries ¶

func ProduceRetries(n int) Opt

ProduceRetries sets the number of tries for producing records, overriding the unlimited default.

If idempotency is enabled (as it is by default), this option is only enforced if it is safe to do so without creating invalid sequence numbers. It is safe to enforce if a record was never issued in a request to Kafka, or if it was requested and received a response.

This option is different from RequestRetries to allow finer grained control of when to fail when producing records.

func RequestRetries ¶

func RequestRetries(n int) Opt

RequestRetries sets the number of tries that retriable requests are allowed, overriding the default of 30. This option does not apply to produce requests; to limit produce request retries, see ProduceRetries.

func RetryBackoff ¶

func RetryBackoff(backoff func(int) time.Duration) Opt

RetryBackoff sets the backoff strategy for how long to backoff for a given amount of retries, overriding the default exponential backoff that ranges from 100ms min to 1s max.

This (roughly) corresponds to Kafka's retry.backoff.ms setting and retry.backoff.max.ms (which is being introduced with KIP-500).

func RetryTimeout ¶

func RetryTimeout(t func(int16) time.Duration) Opt

RetryTimeout sets the upper limit on how long we allow requests to retry, overriding the default of 5m for EndTxn requests, 1m for all others.

This timeout applies to any request issued through a client's Request function. It does not apply to fetches nor produces.

The function is called with the request key that is being retried. While it is not expected that the request key will be used, including it gives users the opportinuty to have different retry timeouts for different keys.

If the function returns zero, there is no retry timeout.

The timeout is evaluated after a request is issued. If a retry backoff places the next request past the retry timeout deadline, the request will still be tried once more once the backoff expires.

func SASL ¶

func SASL(sasls ...sasl.Mechanism) Opt

SASL appends sasl authentication options to use for all connections.

SASL is tried in order; if the broker supports the first mechanism, all connections will use that mechanism. If the first mechanism fails, the client will pick the first supported mechanism. If the broker does not support any client mechanisms, connections will fail.

func SeedBrokers ¶

func SeedBrokers(seeds ...string) Opt

SeedBrokers sets the seed brokers for the client to use, overriding the default 127.0.0.1:9092.

Any seeds that are missing a port use the default Kafka port 9092.

func SoftwareNameAndVersion ¶

func SoftwareNameAndVersion(name, version string) Opt

SoftwareNameAndVersion sets the client software name and version that will be sent to Kafka as part of the ApiVersions request as of Kafka 2.4.0, overriding the default "kgo" and internal version number.

Kafka exposes this through metrics to help operators understand the impact of clients.

It is generally not recommended to set this. As well, if you do, the name and version must match the following regular expression:

[a-zA-Z0-9](?:[a-zA-Z0-9\\-.]*[a-zA-Z0-9])?

Note this means neither the name nor version can be empty.

func WithHooks ¶

func WithHooks(hooks ...Hook) Opt

WithHooks sets hooks to call whenever relevant.

Hooks can be used to layer in metrics (such as Prometheus hooks) or anything else. The client will call all hooks in order. See the Hooks interface for more information, as well as any interface that contains "Hook" in the name to know the available hooks. A single hook can implement zero or all hook interfaces, and only the hooks that it implements will be called.

func WithLogger ¶

func WithLogger(l Logger) Opt

WithLogger sets the client to use the given logger, overriding the default to not use a logger.

It is invalid to use a nil logger; doing so will cause panics.

func StickyKeyPartitioner ¶

func StickyKeyPartitioner(overrideHasher PartitionerHasher) Partitioner

StickyKeyPartitioner mirrors the default Java partitioner from Kafka's 2.4.0 release (see KAFKA-8601).

This is the same "hash the key consistently, if no key, choose random partition" strategy that the Java partitioner has always used, but rather than always choosing a random partition, the partitioner pins a partition to produce to until that partition rolls over to a new batch. Only when rolling to new batches does this partitioner switch partitions.

The benefit with this pinning is less CPU utilization on Kafka brokers. Over time, the random distribution is the same, but the brokers are handling on average larger batches.

overrideHasher is optional; if nil, this will return a partitioner that partitions exactly how Kafka does. Specifically, the partitioner will use murmur2 to hash keys, will mask out the 32nd bit, and then will mod by the number of potential partitions.

func StickyPartitioner ¶

func StickyPartitioner() Partitioner

StickyPartitioner is the same as StickyKeyPartitioner, but with no logic to consistently hash keys. That is, this only partitions according to the sticky partition strategy.

func KafkaHasher ¶

func KafkaHasher(hashFn func([]byte) uint32) PartitionerHasher

KafkaHasher returns a PartitionerHasher using hashFn that mirrors how Kafka partitions after hashing data.

func SaramaHasher ¶

func SaramaHasher(hashFn func([]byte) uint32) PartitionerHasher

SaramaHasher returns a PartitionerHasher using hashFn that mirrors how Sarama partitions after hashing data.

func BatchCompression ¶

func BatchCompression(preference ...CompressionCodec) ProducerOpt

BatchCompression sets the compression codec to use for producing records.

Compression is chosen in the order preferred based on broker support. For example, zstd compression was introduced in Kafka 2.1.0, so the preference can be first zstd, fallback snappy, fallback none.

The default preference is [snappy, none], which should be fine for all old consumers since snappy compression has existed since Kafka 0.8.0. To use zstd, your brokers must be at least 2.1.0 and all consumers must be upgraded to support decoding zstd records.

func BatchMaxBytes ¶

func BatchMaxBytes(v int32) ProducerOpt

BatchMaxBytes upper bounds the size of a record batch, overriding the default 1MB.

This corresponds to Kafka's max.message.bytes, which defaults to 1,000,012 bytes (just over 1MB).

Record batches are independent of a ProduceRequest: a record batch is specific to a topic and partition, whereas the produce request can contain many record batches for many topics.

If a single record encodes larger than this number (before compression), it will will not be written and a callback will have the appropriate error.

Note that this is the maximum size of a record batch before compression. If a batch compresses poorly and actually grows the batch, the uncompressed form will be used.

func DisableIdempotentWrite ¶

func DisableIdempotentWrite() ProducerOpt

DisableIdempotentWrite disables idempotent produce requests, opting out of Kafka server-side deduplication in the face of reissued requests due to transient network problems.

Idempotent production is strictly a win, but does require the IDEMPOTENT_WRITE permission on CLUSTER, and not all clients can have that permission.

This option is incompatible with specifying a transactional id.

func Linger ¶

func Linger(linger time.Duration) ProducerOpt

Linger sets how long individual topic partitions will linger waiting for more records before triggering a request to be built.

Note that this option should only be used in low volume producers. The only benefit of lingering is to potentially build a larger batch to reduce cpu usage on the brokers if you have many producers all producing small amounts.

If a produce request is triggered by any topic partition, all partitions with a possible batch to be sent are used and all lingers are reset.

As mentioned, the linger is specific to topic partition. A high volume producer will likely be producing to many partitions; it is both unnecessary to linger in this case and inefficient because the client will have many timers running (and stopping and restarting) unnecessarily.

func ManualFlushing ¶

func ManualFlushing() ProducerOpt

ManualFlushing disables auto-flushing when producing. While you can still set lingering, it would be useless to do so.

With manual flushing, producing while MaxBufferedRecords have already been produced and not flushed will return ErrMaxBuffered.

func MaxBufferedRecords ¶

func MaxBufferedRecords(n int) ProducerOpt

MaxBufferedRecords sets the max amount of records the client will buffer, blocking produces until records are finished if this limit is reached. This overrides the unbounded default.

func OnDataLoss ¶

func OnDataLoss(fn func(string, int32)) ProducerOpt

OnDataLoss sets a function to call if data loss is detected when producing records if the client is configured to continue on data loss. Thus, this option is mutually exclusive with StopOnDataLoss.

The passed function will be called with the topic and partition that data loss was detected on.

func ProduceRequestTimeout ¶

func ProduceRequestTimeout(limit time.Duration) ProducerOpt

ProduceRequestTimeout sets how long Kafka broker's are allowed to respond to produce requests, overriding the default 30s. If a broker exceeds this duration, it will reply with a request timeout error.

This corresponds to Kafka's request.timeout.ms setting, but only applies to produce requests.

func ProduceTopic ¶

func ProduceTopic(t string) ProducerOpt

ProduceTopic sets the default topic to produce to if the topic field is empty in a kgo.Record.

If this option is not used, if a record has an empty topic, the record cannot be produced and will be failed immediately.

func RecordPartitioner ¶

func RecordPartitioner(partitioner Partitioner) ProducerOpt

RecordPartitioner uses the given partitioner to partition records, overriding the default StickyKeyPartitioner.

func RecordTimeout ¶

func RecordTimeout(timeout time.Duration) ProducerOpt

RecordTimeout sets a rough time of how long a record can sit around in a batch before timing out, overriding the ulimited default.

If idempotency is enabled (as it is by default), this option is only enforced if it is safe to do so without creating invalid sequence numbers. It is safe to enforce if a record was never issued in a request to Kafka, or if it was requested and received a response.

The timeout for all records in a batch inherit the timeout of the first record in that batch. That is, once the first record's timeout expires, all records in the batch are expired. This generally is a non-issue unless using this option with lingering. In that case, simply add the linger to the record timeout to avoid problems.

The timeout is only evaluated evaluated before writing a request or after a produce response. Thus, a sink backoff may delay record timeout slightly.

func RequiredAcks ¶

func RequiredAcks(acks Acks) ProducerOpt

RequiredAcks sets the required acks for produced records, overriding the default RequireAllISRAcks.

func StopOnDataLoss ¶

func StopOnDataLoss() ProducerOpt

StopOnDataLoss sets the client to stop producing if data loss is detected, overriding the default false.

Note that if using this option, it is strongly recommended to not have a retry limit. Doing so may lead to errors where the client fails a batch on a recoverable error, which internally bumps the idempotent sequence number used for producing, which may then later cause an inadvertent out of order sequence number and false "data loss" detection.

func TransactionTimeout ¶

func TransactionTimeout(timeout time.Duration) ProducerOpt

TransactionTimeout sets the allowed for a transaction, overriding the default 40s. It is a good idea to keep this less than a group's session timeout, so that a group member will always be alive for the duration of a transaction even if connectivity dies. This helps prevent a transaction finishing after a rebalance, which is problematic pre-Kafka 2.5.0. If you are on Kafka 2.5.0+, then you can use the RequireStableFetchOffsets option when assigning the group, and you can set this to whatever you would like.

Transaction timeouts begin when the first record is produced within a transaction, not when a transaction begins.

func TransactionalID ¶

func TransactionalID(id string) ProducerOpt

TransactionalID sets a transactional ID for the client, ensuring that records are produced transactionally under this ID (exactly once semantics).

For Kafka-to-Kafka transactions, the transactional ID is only one half of the equation. You must also assign a group to consume from.

To produce transactionally, you first BeginTransaction, then produce records consumed from a group, then you EndTransaction. All records prodcued outside of a transaction will fail immediately with an error.

After producing a batch, you must commit what you consumed. Auto committing offsets is disabled during transactional consuming / producing.

Note that, unless using Kafka 2.5.0, a consumer group rebalance may be problematic. Production should finish and be committed before the client rejoins the group. It may be safer to use an eager group balancer and just abort the transaction. Alternatively, any time a partition is revoked, you could abort the transaction and reset offsets being consumed.

If the client detects an unrecoverable error, all records produced thereafter will fail.

Lastly, the default read level is READ_UNCOMMITTED. Be sure to use the ReadIsolationLevel option if you want to only read committed.

func KeySliceRecord ¶

func KeySliceRecord(key, value []byte) *Record

KeySliceRecord returns a Record with the Key and Value fields set to the input key and value slices. For producing, this function is useful in tandem with the client-level ProduceTopic option.

func KeyStringRecord ¶

func KeyStringRecord(key, value string) *Record

KeyStringRecord returns a Record with the Key and Value fields set to the input key and value strings. For producing, this function is useful in tandem with the client-level ProduceTopic option.

This function uses the 'unsafe' package to avoid copying value into a slice.

NOTE: It is NOT SAFE to modify the record's value. This function should only be used if you only ever read record fields. This function can safely be used for producing; the client never modifies a record's key nor value fields.

func SliceRecord ¶

func SliceRecord(value []byte) *Record

SliceRecord returns a Record with the Value field set to the input value slice. For producing, this function is useful in tandem with the client-level ProduceTopic option.

func StringRecord ¶

func StringRecord(value string) *Record

StringRecord returns a Record with the Value field set to the input value string. For producing, this function is useful in tandem with the client-level ProduceTopic option.

This function uses the 'unsafe' package to avoid copying value into a slice.

NOTE: It is NOT SAFE to modify the record's value. This function should only be used if you only ever read record fields. This function can safely be used for producing; the client never modifies a record's key nor value fields.