config

package
v0.0.0-...-b937326 Latest Latest
Warning

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

 Go to latest Published: Oct 27, 2020 License: Apache-2.0 Imports: 6 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type ConfigHighLevel 

type ConfigHighLevel struct {
	// Initial list of brokers as a CSV list of broker host or host:port. The application may also use rd_kafka_brokers_add() to add brokers during runtime.
	// kafka broker 地址
	// 对应为 metadata.broker.list 或 bootstrap.servers
	// alias: bootstrap.servers
	MetadataBrokerList string `json:"metadata.broker.list"`

	// librdkafka statistics emit interval. The application also needs to register a stats callback using rd_kafka_conf_set_stats_cb(). The granularity is 1000ms. A value of 0 disables statistics.
	// range: 0 ~ 86400e3
	// Type: integer
	StatisticsIntervalMs int `json:"statistics.interval.ms"`

	// high	Request broker's supported API versions to adjust functionality to available protocol features. If set to false, or the ApiVersionRequest fails, the fallback version broker.version.fallback will be used. NOTE: Depends on broker version >=0.10.0. If the request is not supported by (an older) broker the broker.version.fallback fallback is used.
	// Type: boolean
	ApiVersionRequest bool `json:"api.version.request"`

	// Protocol used to communicate with brokers.
	// range plaintext, ssl, sasl_plaintext, sasl_ssl
	// Type: enum value
	SecurityProtocol string `json:"security.protocol"`

	// SASL mechanism to use for authentication. Supported: GSSAPI, PLAIN, SCRAM-SHA-256, SCRAM-SHA-512, OAUTHBEARER. NOTE: Despite the name only one mechanism must be configured.
	// Type: string
	// alias: sasl.mechanism
	SaslMechanisms string `json:"sasl.mechanisms"`

	// SASL username for use with the PLAIN and SASL-SCRAM-.. mechanisms
	// Type: string
	SaslUsername string `json:"sasl.username"`

	// SASL password for use with the PLAIN and SASL-SCRAM-.. mechanism
	// Type: string
	SaslPassword string `json:"sasl.password"`
}

func DefaultConfigHigh 

func DefaultConfigHigh() ConfigHighLevel

type ConfigLowLevel 

type ConfigLowLevel struct {
	// Indicates the builtin features for this build of librdkafka. An application can either query this value or attempt to set it with its list of required features to check for library support.
	// Type: CSV flags
	BuiltInFeatures string `json:"builtin.features"`

	// client identifier
	ClientID string `json:"client.id"`

	// Maximum size for message to be copied to buffer. Messages larger than this will be passed by reference (zero-copy) at the expense of larger iovecs.
	// range: 0 ~ 1e9
	// Type: integer
	MessageCopyMaxBytes int `json:"message.copy.max.bytes"`

	// Maximum number of in-flight requests per broker connection. This is a generic property applied to all broker communication, however it is primarily relevant to produce requests. In particular, note that other mechanisms limit the number of outstanding consumer fetch request per broker to one.
	// range: 1 ~ 1e6
	// Type: integer
	// alias: max.in.flight
	MaxInFlightRequestsPerConnection int `json:"max.in.flight.requests.per.connection"`

	// Non-topic request timeout in milliseconds. This is for metadata requests, etc.
	// range: 10 ~ 9e5
	// Type: integer
	MetadataRequestTimeoutMs int `json:"metadata.request.timeout.ms"`

	// Period of time in milliseconds at which topic and broker metadata is refreshed in order to proactively discover any new brokers, topics, partitions or partition leader changes. Use -1 to disable the intervalled refresh (not recommended). If there are no locally referenced topics (no topic objects created, no messages produced, no subscription or no assignment) then only the broker list will be refreshed every interval but no more often than every 10s.
	// range: -1 ~ 3.6e6
	// Type: integer
	TopicMetadataRefreshIntervalMs int `json:"topic.metadata.refresh.interval.ms"`

	// Metadata cache max age. Defaults to topic.metadata.refresh.interval.ms * 3
	// range: 1 ~ 86400e3
	// Type: integer
	MetadataMaxAgeMs int `json:"metadata.max.age.ms"`

	// When a topic loses its leader a new metadata request will be enqueued with this initial interval, exponentially increasing until the topic metadata has been refreshed. This is used to recover quickly from transitioning leader brokers.
	// range 1 ~ 6e5
	// Type: integer
	TopicMetadataRefreshFastIntervalMs int `json:"topic.metadata.refresh.fast.interval.ms"`

	// Sparse metadata requests (consumes less network bandwidth)
	// Type: boolean
	TopicMetadataRefreshSparse bool `json:"topic.metadata.refresh.sparse"`

	// Topic blacklist, a comma-separated list of regular expressions for matching topic names that should be ignored in broker metadata information as if the topics did not exist.
	// Type: pattern list
	TopicBlacklist string `json:"topic.blacklist"`
	// Default timeout for network requests. Producer: ProduceRequests will use the lesser value of socket.timeout.ms and remaining message.timeout.ms for the first message in the batch. Consumer: FetchRequests will use fetch.wait.max.ms + socket.timeout.ms. Admin: Admin requests will use socket.timeout.ms or explicitly set rd_kafka_AdminOptions_set_operation_timeout() value.
	// range 10 ~ 3e5
	// Type: integer
	SocketTimeoutMs int `json:"socket.timeout.ms"`

	// Broker socket send buffer size. System default is used if 0.
	// range: 0 ~ 1e8
	// Type: integer
	SocketSendBufferBytes int `json:"socket.send.buffer.bytes"`

	// Broker socket receive buffer size. System default is used if 0.
	// range: 0 ~ 1e8
	// Type: integer
	SocketReceiveBufferBytes int `json:"socket.receive.buffer.bytes"`

	// Enable TCP keep-alives (SO_KEEPALIVE) on broker sockets
	// Type: boolean
	SocketKeepaliveEnable bool `json:"socket.keepalive.enable"`

	// Disable the Nagle algorithm (TCP_NODELAY) on broker sockets.
	// Type: boolean
	SocketNagleDisable bool `json:"socket.nagle.disable"`

	// Disconnect from broker when this number of send failures (e.g., timed out requests) is reached. Disable with 0. WARNING: It is highly recommended to leave this setting at its default value of 1 to avoid the client and broker to become desynchronized in case of request timeouts. NOTE: The connection is automatically re-established.
	// range 0 ~ 1e6
	// Type: integer
	SocketMaxFails int `json:"socket.max.fails"`

	// How long to cache the broker address resolving results (milliseconds).
	// range 0 ~ 86400e3
	// Type: integer
	BrockerAddressTTL int `json:"broker.address.ttl"`

	// Allowed broker IP address families: any, v4, v6
	// range: any, v4, v6
	// Type: enum value
	BrockerAddressFamily string `json:"broker.address.family"`
	// See rd_kafka_conf_set_events()
	// range: 0 ~ 2^31-1
	// Type: integer
	EnabledEvents int `json:"enabled_events"`

	// Error callback (set with rd_kafka_conf_set_error_cb())
	// Type: see dedicated API
	ErrorCb interface{} `json:"-"`

	// Throttle callback (set with rd_kafka_conf_set_throttle_cb())
	// Type: see dedicated API
	ThrottleCb interface{} `json:"-"`

	// Statistics callback (set with rd_kafka_conf_set_stats_cb())
	// Type: see dedicated API
	StatsCb interface{} `json:"-"`

	// Log callback (set with rd_kafka_conf_set_log_cb())
	// Type: see dedicated API
	LogCb interface{} `json:"-"`

	// Logging level (syslog(3) levels)
	// range: 0 ~ 7
	// Type: integer
	LogLevel int `json:"log_level"`

	// Disable spontaneous log_cb from internal librdkafka threads, instead enqueue log messages on queue set with rd_kafka_set_log_queue() and serve log callbacks or events through the standard poll APIs. NOTE: Log messages will linger in a temporary queue until the log queue has been set.
	// Type: boolean
	LogQueue bool `json:"log.queue"`

	// Print internal thread name in log messages (useful for debugging librdkafka internals)
	// Type: boolean
	LogThreadName bool `json:"log.thread.name"`

	// If enabled librdkafka will initialize the POSIX PRNG with srand(current_time.milliseconds) on the first invocation of rd_kafka_new(). If disabled the application must call srand() prior to calling rd_kafka_new().
	// Type: boolean
	EnableRandomSeed bool `json:"enable.random.seed"`

	// Log broker disconnects. It might be useful to turn this off when interacting with 0.9 brokers with an aggressive connection.max.idle.ms value.
	// Type: boolean
	LogConnectionClose bool `json:"log.connection.close"`

	// Background queue event callback (set with rd_kafka_conf_set_background_event_cb())
	// Type: see dedicated API
	BackgroundEventCb interface{} `json:"-"`

	// Socket creation callback to provide race-free CLOEXEC
	// Type: see dedicated API
	SocketCb interface{} `json:"-"`

	// 	Socket connect callback
	// Type: see dedicated API
	ConnectCb interface{} `json:"-"`

	// Socket close callback
	// Type: see dedicated API
	ClosesocketCb interface{} `json:"-"`

	// 	File open callback to provide race-free CLOEXEC
	// Type: see dedicated API
	OpenCb interface{} `json:"-"`

	// Application opaque (set with rd_kafka_conf_set_opaque())
	// Type: see dedicated API
	Opaque interface{} `json:"-"`

	// Default topic configuration for automatically subscribed topics
	// Type: see dedicated API
	DefaultTopicConf interface{} `json:"-"`

	// Signal that librdkafka will use to quickly terminate on rd_kafka_destroy(). If this signal is not set then there will be a delay before rd_kafka_wait_destroyed() returns true as internal threads are timing out their system calls. If this signal is set however the delay will be minimal. The application should mask this signal as an internal signal handler is installed.
	// range 0 ~ 128
	// Type: integer
	InternalTerminaltionSignal int `json:"internal.termination.signal"`

	// Timeout for broker API version requests.
	// range 1 ~ 3e5
	// Type: integer
	ApiVersionRequestTimeoutMs int `json:"api.version.request.timeout.ms"`

	// A cipher suite is a named combination of authentication, encryption, MAC and key exchange algorithm used to negotiate the security settings for a network connection using TLS or SSL network protocol. See manual page for ciphers(1) and `SSL_CTX_set_cipher_list(3).
	// Type: string
	SslCipherSuites string `json:"ssl.cipher.suites"`

	// The supported-curves extension in the TLS ClientHello message specifies the curves (standard/named, or 'explicit' GF(2^k) or GF(p)) the client is willing to have the server use. See manual page for SSL_CTX_set1_curves_list(3). OpenSSL >= 1.0.2 required.
	// Type: string
	SslCurvesList string `json:"ssl.curves.list"`

	// The client uses the TLS ClientHello signature_algorithms extension to indicate to the server which signature/hash algorithm pairs may be used in digital signatures. See manual page for SSL_CTX_set1_sigalgs_list(3). OpenSSL >= 1.0.2 required.
	// Type: string
	SslSigalgsList string `json:"ssl.sigalgs.list"`

	// Path to client's private key (PEM) used for authentication.
	// Type: string
	SslKeyLocation string `json:"ssl.key.location"`

	// Private key passphrase (for use with ssl.key.location and set_ssl_cert())
	// Type: string
	SslKeyPassword string `json:"ssl.key.password"`

	// Client's private key string (PEM format) used for authentication.
	// Type: string
	SslKeyPem string `json:"ssl.key.pem"`

	// 	Client's private key as set by rd_kafka_conf_set_ssl_cert()
	// Type: see dedicated API
	SslKey interface{} `json:"-"`

	// Path to client's public key (PEM) used for authentication.
	// Type: string
	SslCertificateLocaltion string `json:"ssl.certificate.location"`

	// Client's public key string (PEM format) used for authentication.
	// Type: string
	SslCertificatePem string `json:"ssl.certificate.pem"`

	// Client's public key as set by rd_kafka_conf_set_ssl_cert()
	// Type: see dedicated API
	SslCertificate interface{} `json:"-"`

	// File or directory path to CA certificate(s) for verifying the broker's key. Defaults: On Windows the system's CA certificates are automatically looked up in the Windows Root certificate store. On Mac OSX it is recommended to install openssl using Homebrew, to provide CA certificates. On Linux install the distribution's ca-certificates package. If OpenSSL is statically linked or ssl.ca.location is set to probe a list of standard paths will be probed and the first one found will be used as the default CA certificate location path. If OpenSSL is dynamically linked the OpenSSL library's default path will be used (see OPENSSLDIR in openssl version -a).
	// Type: string
	SslCaLocation string `json:"ssl.ca.location"`

	// 	CA certificate as set by rd_kafka_conf_set_ssl_cert()
	// Type: see dedicated API
	SslCa interface{} `json:"-"`

	// Path to CRL for verifying broker's certificate validity.
	// Type: string
	SslCrlLocation string `json:"ssl.crl.location"`

	// Path to client's keystore (PKCS#12) used for authentication.
	// Type: string
	SslKeystoreLocation string `json:"ssl.keystore.location"`

	// Client's keystore (PKCS#12) password.
	// Type: string
	SslKeystorePassword string `json:"ssl.keystore.password"`

	// Enable OpenSSL's builtin broker (server) certificate verification. This verification can be extended by the application by implementing a certificate_verify_cb.
	// Type: boolean
	EnableSslCertificateVerification bool `json:"enable.ssl.certificate.verification"`

	// Endpoint identification algorithm to validate broker hostname using broker certificate. https - Server (broker) hostname verification as specified in RFC2818. none - No endpoint verification. OpenSSL >= 1.0.2 required.
	// range: none, https
	// Type: enum value
	SslEndpointIdentificationAlgorithm string `json:"ssl.endpoint.identification.algorithm"`

	// 	Callback to verify the broker certificate chain.
	// Type: see dedicated API
	SslCertificateVerifyCb interface{} `json:"-"`

	// Kerberos principal name that Kafka runs as, not including /hostname@REALM
	// Type: string
	SaslKerberosServiceName string `json:"sasl.kerberos.service.name"`

	// This client's Kerberos principal name. (Not supported on Windows, will use the logon user's principal).
	// Type: string
	SaslKerberosPrincipal string `json:"sasl.kerberos.principal"`

	// Shell command to refresh or acquire the client's Kerberos ticket. This command is executed on client creation and every sasl.kerberos.min.time.before.relogin (0=disable). %{config.prop.name} is replaced by corresponding config object value.
	// Type: string
	SaslKerberosKinitCmd string `json:"sasl.kerberos.kinit.cmd"`

	// Path to Kerberos keytab file. This configuration property is only used as a variable in sasl.kerberos.kinit.cmd as ... -t "%{sasl.kerberos.keytab}".
	// Type: string
	SaslKerberosKeytab string `json:"sasl.kerberos.keytab"`

	// Minimum time in milliseconds between key refresh attempts. Disable automatic key refresh by setting this property to 0.
	// range: 0 ~ 86400e3
	// Type: integer
	SaslKerberosMinTimeBeforeRelogin int `json:"sasl.kerberos.min.time.before.relogin"`

	// SASL/OAUTHBEARER configuration. The format is implementation-dependent and must be parsed accordingly. The default unsecured token implementation (see https://tools.ietf.org/html/rfc7515#appendix-A.5) recognizes space-separated name=value pairs with valid names including principalClaimName, principal, scopeClaimName, scope, and lifeSeconds. The default value for principalClaimName is "sub", the default value for scopeClaimName is "scope", and the default value for lifeSeconds is 3600. The scope value is CSV format with the default value being no/empty scope. For example: principalClaimName=azp principal=admin scopeClaimName=roles scope=role1,role2 lifeSeconds=600. In addition, SASL extensions can be communicated to the broker via extension_NAME=value. For example: principal=admin extension_traceId=123
	// Type: string
	SaslOauthbearerConfig string `json:"sasl.oauthbearer.config"`

	// Enable the builtin unsecure JWT OAUTHBEARER token handler if no oauthbearer_refresh_cb has been set. This builtin handler should only be used for development or testing, and not in production.
	// Type: boolean
	EnableSaslOauthbearerUnsecureJwt bool `json:"enable.sasl.oauthbearer.unsecure.jwt"`

	// SASL/OAUTHBEARER token refresh callback (set with rd_kafka_conf_set_oauthbearer_token_refresh_cb(), triggered by rd_kafka_poll(), et.al. This callback will be triggered when it is time to refresh the client's OAUTHBEARER token.
	// Type: see dedicated API
	OauthbearerTokenRefreshCb interface{} `json:"-"`

	// List of plugin libraries to load (; separated). The library search path is platform dependent (see dlopen(3) for Unix and LoadLibrary() for Windows). If no filename extension is specified the platform-specific extension (such as .dll or .so) will be appended automatically.
	// Type: string
	PluginLibraryPaths string `json:"plugin.library.paths"`

	// Interceptors added through rd_kafka_conf_interceptor_add_..() and any configuration handled by interceptors.
	// Type: see dedicated API
	Interceptors interface{} `json:"-"`
}

func DefaultConfigLow 

func DefaultConfigLow() ConfigLowLevel

type ConfigMediumLevel 

type ConfigMediumLevel struct {
	// Maximum Kafka protocol request message size. Due to differing framing overhead between protocol versions the producer is unable to reliably enforce a strict max message limit at produce time and may exceed the maximum size by one message in protocol ProduceRequests, the broker will enforce the the topic's max.message.bytes limit (see Apache Kafka documentation).
	// range: 1e3 ~ 1e9
	// Type: integer
	MessageMaxBytes int `json:"message.max.bytes"`

	// Maximum Kafka protocol response message size. This serves as a safety precaution to avoid memory exhaustion in case of protocol hickups. This value must be at least fetch.max.bytes + 512 to allow for protocol overhead; the value is adjusted automatically unless the configuration property is explicitly set.
	// range: 1e3 ~ 2^31-1
	// Type: integer
	ReceiveMessageMaxBytes int `json:"receive.message.max.bytes"`

	// A comma-separated list of debug contexts to enable. Detailed Producer debugging: broker,topic,msg. Consumer: consumer,cgrp,topic,fetch
	// range: generic, broker, topic, metadata, feature, queue, msg, protocol, cgrp, security, fetch, interceptor, plugin, consumer, admin, eos, mock, all
	// Type: CSV flags
	Debug string `json:"debug"`

	// The initial time to wait before reconnecting to a broker after the connection has been closed. The time is increased exponentially until reconnect.backoff.max.ms is reached. -25% to +50% jitter is applied to each reconnect backoff. A value of 0 disables the backoff and reconnects immediately.
	// range 0 ~ 3.6e6
	// Type: integer
	ReconnectBackoffMs int `json:"reconnect.backoff.ms"`

	// The maximum time to wait before reconnecting to a broker after the connection has been closed.
	// range 0 ~ 3.6e6
	// Type: integer
	ReconnectBackoffMaxMs int `json:"reconnect.backoff.max.ms"`

	// Dictates how long the broker.version.fallback fallback is used in the case the ApiVersionRequest fails. NOTE: The ApiVersionRequest is only issued when a new connection to the broker is made (such as after an upgrade).
	// range 0 ~ 6048e5
	// Type: integer
	ApiVersionFallbackMs int `json:"api.version.fallback.ms"`

	// Older broker versions (before 0.10.0) provide no way for a client to query for supported protocol features (ApiVersionRequest, see api.version.request) making it impossible for the client to know what features it may use. As a workaround a user may set this property to the expected broker version and the client will automatically adjust its feature set accordingly if the ApiVersionRequest fails (or is disabled). The fallback broker version will be used for api.version.fallback.ms. Valid values are: 0.9.0, 0.8.2, 0.8.1, 0.8.0. Any other value >= 0.10, such as 0.10.2.1, enables ApiVersionRequests.
	// Type: string
	BrockerVersionFallback string `json:"broker.version.fallback"`
}

func DefaultConfigMedium 

func DefaultConfigMedium() ConfigMediumLevel

type Consumer 

type Consumer struct {
	*ConsumerConfig
	*kafka.Consumer
}

type ConsumerConfig 

type ConsumerConfig struct {
	KafkaConfig kafkaConsumerConfig `json:"kafka_config"`
	// contains filtered or unexported fields
}

func DefaultConsumerConfig 

func DefaultConsumerConfig() *ConsumerConfig

func StdConsumerConfig 

func StdConsumerConfig(path string) *ConsumerConfig

StdConsumerConfig ...

func (*ConsumerConfig) BuildConsumer 

func (config *ConsumerConfig) BuildConsumer() *Consumer

Build ...

type ConsumerConfigHighLevel 

type ConsumerConfigHighLevel struct {
	ConfigHighLevel `json:"config_high_level,flatten"`

	// Client group id string. All clients sharing the same group.id belong to the same group.
	// Type: string
	GroupID string `json:"group.id"`

	// Client group session and failure detection timeout. The consumer sends periodic heartbeats (heartbeat.interval.ms) to indicate its liveness to the broker. If no hearts are received by the broker for a group member within the session timeout, the broker will remove the consumer from the group and trigger a rebalance. The allowed range is configured with the broker configuration properties group.min.session.timeout.ms and group.max.session.timeout.ms. Also see max.poll.interval.ms.
	// range: 1 ~ 3.6e6
	// Type: integer
	SessionTimeoutMs int `json:"session.timeout.ms"`

	// Maximum allowed time between calls to consume messages (e.g., rd_kafka_consumer_poll()) for high-level consumers. If this interval is exceeded the consumer is considered failed and the group will rebalance in order to reassign the partitions to another consumer group member. Warning: Offset commits may be not possible at this point. Note: It is recommended to set enable.auto.offset.store=false for long-time processing applications and then explicitly store offsets (using offsets_store()) after message processing, to make sure offsets are not auto-committed prior to processing has finished. The interval is checked two times per second. See KIP-62 for more information.
	// range: 1 ~ 86400e3
	// Type: integer
	MaxPollIntervalMs int `json:"max.poll.interval.ms"`

	// Automatically and periodically commit offsets in the background. Note: setting this to false does not prevent the consumer from fetching previously committed start offsets. To circumvent this behaviour set specific start offsets per partition in the call to assign().
	// Type: boolean
	EnableAutoCommit bool `json:"enable.auto.commit"`

	// Automatically store offset of last message provided to application. The offset store is an in-memory store of the next offset to (auto-)commit for each partition.
	// Type: boolean
	EnableAutoOffsetStore bool `json:"enable.auto.offset.store"`

	// Controls how to read messages written transactionally: read_committed - only return transactional messages which have been committed. read_uncommitted - return all messages, even transactional messages which have been aborted.
	// range: read_uncommitted, read_committed
	// Type: enum value
	IsolationLevel string `json:"isolation.level"`
}

func DefaultConsumerConfigHigh 

func DefaultConsumerConfigHigh() ConsumerConfigHighLevel

type ConsumerConfigLowLevel 

type ConsumerConfigLowLevel struct {
	ConfigLowLevel `json:"config_low_level,flatten"`

	// Group session keepalive heartbeat interval.
	// range: 1 ~ 3.6e6
	// Type: integer
	HeartbeatIntervalMs int `json:"heartbeat.interval.ms"`

	// Group protocol type
	// Type: string
	GroupProtocolType string `json:"group.protocol.type"`

	// How often to query for the current client group coordinator. If the currently assigned coordinator is down the configured query interval will be divided by ten to more quickly recover in case of coordinator reassignment.
	// range: 1 ~ 3.6e6
	// Type: integer
	CoordinatorQueryIntervalMs int `json:"coordinator.query.interval.ms"`

	// Maximum time the broker may wait to fill the Fetch response with fetch.min.bytes of messages.
	// range: 0 ~ 3e6
	// Type: integer
	FetchWaitMaxMs int `json:"fetch.wait.max.ms"`

	// Minimum number of bytes the broker responds with. If fetch.wait.max.ms expires the accumulated data will be sent to the client regardless of this setting.
	// range: 1 ~ 1e8
	// Type: integer
	FetchMinBytes int `json:"fetch.min.bytes"`

	// Message consume callback (set with rd_kafka_conf_set_consume_cb())
	// Type: see dedicated API
	ConsumeCb interface{} `json:"-"`

	// Called after consumer group has been rebalanced (set with rd_kafka_conf_set_rebalance_cb())
	// Type: see dedicated API
	RebalanceCb interface{} `json:"-"`

	// Offset commit result propagation callback. (set with rd_kafka_conf_set_offset_commit_cb())
	// Type: see dedicated API
	OffsetCommitCb interface{} `json:"-"`

	// Emit RD_KAFKA_RESP_ERR__PARTITION_EOF event whenever the consumer reaches the end of a partition.
	// Type: boolean
	EnablePartitionEof bool `json:"enable.partition.eof"`

	// A rack identifier for this client. This can be any string value which indicates where this client is physically located. It corresponds with the broker config broker.rack.
	// Type: string
	ClientRack string `json:"client.rack"`
}

func DefaultConsumerConfigLow 

func DefaultConsumerConfigLow() ConsumerConfigLowLevel

type ConsumerConfigMediumLevel 

type ConsumerConfigMediumLevel struct {
	ConfigMediumLevel `json:"config_medium_level,flatten"`

	// Enable static group membership. Static group members are able to leave and rejoin a group within the configured session.timeout.ms without prompting a group rebalance. This should be used in combination with a larger session.timeout.ms to avoid group rebalances caused by transient unavailability (e.g. process restarts). Requires broker version >= 2.3.0.
	// Type: string
	GroupInstanceId string `json:"group.instance.id"`

	// Name of partition assignment strategy to use when elected group leader assigns partitions to group members.
	// Type: string
	PartitionAssignmentStrategy string `json:"partition.assignment.strategy"`

	// The frequency in milliseconds that the consumer offsets are committed (written) to offset storage. (0 = disable). This setting is used by the high-level consumer.
	// range: 0 ~ 86400e3
	// Type: integer
	AutoCommitIntervalMs int `json:"auto.commit.interval.ms"`

	// Minimum number of messages per topic+partition librdkafka tries to maintain in the local consumer queue.
	// range 1 ~ 1e7
	// Type: integer
	QueuedMinMessages int `json:"queued.min.messages"`

	// Maximum number of kilobytes of queued pre-fetched messages in the local consumer queue. If using the high-level consumer this setting applies to the single consumer queue, regardless of the number of partitions. When using the legacy simple consumer or when separate partition queues are used this setting applies per partition. This value may be overshot by fetch.message.max.bytes. This property has higher priority than queued.min.messages.
	// range: 1 ~ 2097151
	// Type: integer
	QueuedMaxMessagesKbytes int `json:"queued.max.messages.kbytes"`

	// Maximum time the broker may wait to fill the Fetch response with fetch.min.bytes of messages.
	// range: 1 ~ 1e9
	// Type: integer
	// alias: max.partition.fetch.bytes
	FetchMessageMaxBytes int `json:"fetch.message.max.bytes"`

	// Maximum amount of data the broker shall return for a Fetch request. Messages are fetched in batches by the consumer and if the first message batch in the first non-empty partition of the Fetch request is larger than this value, then the message batch will still be returned to ensure the consumer can make progress. The maximum message batch size accepted by the broker is defined via message.max.bytes (broker config) or max.message.bytes (broker topic config). fetch.max.bytes is automatically adjusted upwards to be at least message.max.bytes (consumer config).
	// range: 0 ~ 2147483135
	// Type: integer
	FetchMaxBytes int `json:"fetch.max.bytes"`

	// How long to postpone the next fetch request for a topic+partition in case of a fetch error.
	// range: 0 ~ 3e6
	// Type: integer
	FetchErrorBackoffMs int `json:"fetch.error.backoff.ms"`

	// Verify CRC32 of consumed messages, ensuring no on-the-wire or on-disk corruption to the messages occurred. This check comes at slightly increased CPU usage.
	// Type: boolean
	CheckCrcs bool `json:"check.crcs"`
}

func DefaultConsumerConfigMedium 

func DefaultConsumerConfigMedium() ConsumerConfigMediumLevel

type ConsumerTopicConfigHighLevel 

type ConsumerTopicConfigHighLevel struct {
	// Action to take when there is no initial offset in offset store or the desired offset is out of range: 'smallest','earliest' - automatically reset the offset to the smallest offset, 'largest','latest' - automatically reset the offset to the largest offset, 'error' - trigger an error which is retrieved by consuming messages and checking 'message->err'.
	// smallest, earliest, beginning, largest, latest, end, error
	// Type: enum value
	AutoOffsetReset string `json:"auto.offset.reset"`
}

func DefaultConsumerTopicConfigHigh 

func DefaultConsumerTopicConfigHigh() ConsumerTopicConfigHighLevel

type ConsumerTopicConfigLowLevel 

type ConsumerTopicConfigLowLevel struct {
	// Maximum number of messages to dispatch in one rd_kafka_consume_callback*() call (0 = unlimited)
	// range: 0 ~ 1e6
	// Type: integer
	ConsumeCallbackMaxMessages int `json:"consume.callback.max.messages"`
}

func DefaultConsumerTopicConfigLow 

func DefaultConsumerTopicConfigLow() ConsumerTopicConfigLowLevel

type Producer 

type Producer struct {
	*ProducerConfig
	*kafka.Producer
}

func (*Producer) ProduceTo 

func (p *Producer) ProduceTo(topic string, partition int32, offset kafka.Offset, data []byte) error

func (*Producer) ReadProducedEvent 

func (p *Producer) ReadProducedEvent()

func (*Producer) RunMonitor 

func (p *Producer) RunMonitor()

type ProducerConfig 

type ProducerConfig struct {
	KafkaConfig kafkaProducerConfig `json:"kafka_config"`
	// contains filtered or unexported fields
}

func DefaultProducerConfig 

func DefaultProducerConfig() *ProducerConfig

func StdProducerConfig 

func StdProducerConfig(path string) *ProducerConfig

StdKafkaConfig ...

func (*ProducerConfig) BuildProducer 

func (config *ProducerConfig) BuildProducer() *Producer

Build ...

type ProducerConfigHighLevel 

type ProducerConfigHighLevel struct {
	ConfigHighLevel `json:"config_high_level,flatten"`

	// Enables the transactional producer. The transactional.id is used to identify the same transactional producer instance across process restarts. It allows the producer to guarantee that transactions corresponding to earlier instances of the same producer have been finalized prior to starting any new transactions, and that any zombie instances are fenced off. If no transactional.id is provided, then the producer is limited to idempotent delivery (if enable.idempotence is set). Requires broker version >= 0.11.0.
	// Type: string
	TransactionalID string `json:"transactional.id"`

	// When set to true, the producer will ensure that messages are successfully produced exactly once and in the original produce order. The following configuration properties are adjusted automatically (if not modified by the user) when idempotence is enabled: max.in.flight.requests.per.connection=5 (must be less than or equal to 5), retries=INT32_MAX (must be greater than 0), acks=all, queuing.strategy=fifo. Producer instantation will fail if user-supplied configuration is incompatible.
	// Type: boolean
	EnableIdempotence bool `json:"enable.idempotence"`

	// Maximum number of messages allowed on the producer queue. This queue is shared by all topics and partitions.
	// range: 1 ~ 1e7
	// Type: integer
	QueueBufferingMaxMessages int `json:"queue.buffering.max.messages"`

	// Maximum total message size sum allowed on the producer queue. This queue is shared by all topics and partitions. This property has higher priority than queue.buffering.max.messages.
	// range 1 ~ 2147483647
	// Type: integer
	QueueBufferingMaxKbytes int `json:"queue.buffering.max.kbytes"`

	// Delay in milliseconds to wait for messages in the producer queue to accumulate before constructing message batches (MessageSets) to transmit to brokers. A higher value allows larger and more effective (less overhead, improved compression) batches of messages to accumulate at the expense of increased message delivery latency.
	// range: 0 ~ 9e5
	// Type: float; 实际上应该是 integer
	// alias: linger.ms
	QueueBufferingMaxMs int `json:"queue.buffering.max.ms"`

	// How many times to retry sending a failing Message. Note: retrying may cause reordering unless enable.idempotence is set to true.
	// range: 0 ~ 10000000
	// Type: integer
	// alias: retries
	MessageSendMaxRetries int `json:"message.send.max.retries"`
}

func DefaultProducerConfigHigh 

func DefaultProducerConfigHigh() ProducerConfigHighLevel

type ProducerConfigLowLevel 

type ProducerConfigLowLevel struct {
	ConfigLowLevel `json:"config_low_level,flatten"`

	// EXPERIMENTAL: subject to change or removal. When set to true, any error that could result in a gap in the produced message series when a batch of messages fails, will raise a fatal error (ERR__GAPLESS_GUARANTEE) and stop the producer. Messages failing due to message.timeout.ms are not covered by this guarantee. Requires enable.idempotence=true.
	// Type: boolean
	EnableGaplessGuarantee bool `json:"enable.gapless.guarantee"`

	// The threshold of outstanding not yet transmitted broker requests needed to backpressure the producer's message accumulator. If the number of not yet transmitted requests equals or exceeds this number, produce request creation that would have otherwise been triggered (for example, in accordance with linger.ms) will be delayed. A lower number yields larger and more effective batches. A higher value can improve latency when using compression on slow machines.
	// range: 1 ~ 1e6
	// Type: integer
	QueueBufferingBackpressureThreshold int `json:"queue.buffering.backpressure.threshold"`

	// 	Only provide delivery reports for failed messages.
	// Type: boolean
	DeliveryReportOnlyError bool `json:"delivery.report.only.error"`

	// Delivery report callback (set with rd_kafka_conf_set_dr_cb())
	// Type: see dedicated API
	DrCb interface{} `json:"-"`

	// Delivery report callback (set with rd_kafka_conf_set_dr_msg_cb())
	// Type: see dedicated API
	DrMsgCb interface{} `json:"-"`
}

func DefaultProducerConfigLow 

func DefaultProducerConfigLow() ProducerConfigLowLevel

type ProducerConfigMediumLevel 

type ProducerConfigMediumLevel struct {
	ConfigMediumLevel `json:"config_medium_level,flatten"`

	// The maximum amount of time in milliseconds that the transaction coordinator will wait for a transaction status update from the producer before proactively aborting the ongoing transaction. If this value is larger than the transaction.max.timeout.ms setting in the broker, the init_transactions() call will fail with ERR_INVALID_TRANSACTION_TIMEOUT. The transaction timeout automatically adjusts message.timeout.ms and socket.timeout.ms, unless explicitly configured in which case they must not exceed the transaction timeout (socket.timeout.ms must be at least 100ms lower than transaction.timeout.ms). This is also the default timeout value if no timeout (-1) is supplied to the transactional API methods.
	// range: 1e3 ~ 2147483647
	// Type: integer
	TransactionTimeoutMs int `json:"transaction.timeout.ms"`

	// The backoff time in milliseconds before retrying a protocol request.
	// range: 1 ~ 3e5
	// Type: integer
	RetryBackoffMs int `json:"retry.backoff.ms"`

	// compression codec to use for compressing message sets. This is the default value for all topics, may be overridden by the topic configuration property compression.codec.
	// range: none, gzip, snappy, lz4, zstd
	// Type: enum value
	// alias: compression.type
	CompressionCodec string `json:"compression.codec"`

	// Maximum number of messages batched in one MessageSet. The total MessageSet size is also limited by batch.size and message.max.bytes.
	// range: 1 ~ 1e6
	// Type: integer
	BatchNumMessages int `json:"batch.num.messages"`
}

func DefaultProducerConfigMedium 

func DefaultProducerConfigMedium() ProducerConfigMediumLevel

type ProducerTopicConfigHighLevel 

type ProducerTopicConfigHighLevel struct {
	// This field indicates the number of acknowledgements the leader broker must receive from ISR brokers before responding to the request: 0=Broker does not send any response/ack to client, -1 or all=Broker will block until message is committed by all in sync replicas (ISRs). If there are less than min.insync.replicas (broker configuration) in the ISR set the produce request will fail.
	// range: -1 ~ 1e3
	// Type: integer
	// alias: acks
	RequestRequiredAcks int `json:"request.required.acks"`

	// Local message timeout. This value is only enforced locally and limits the time a produced message waits for successful delivery. A time of 0 is infinite. This is the maximum time librdkafka may use to deliver a message (including retries). Delivery error occurs when either the retry count or the message timeout are exceeded. The message timeout is automatically adjusted to transaction.timeout.ms if transactional.id is configured.
	// range: 0 ~ 2147483647
	// Type: integer
	// alias: delivery.timeout.ms
	MessageTimeoutMs int `json:"message.timeout.ms"`

	// Partitioner: random - random distribution, consistent - CRC32 hash of key (Empty and NULL keys are mapped to single partition), consistent_random - CRC32 hash of key (Empty and NULL keys are randomly partitioned), murmur2 - Java Producer compatible Murmur2 hash of key (NULL keys are mapped to single partition), murmur2_random - Java Producer compatible Murmur2 hash of key (NULL keys are randomly partitioned. This is functionally equivalent to the default partitioner in the Java Producer.), fnv1a - FNV-1a hash of key (NULL keys are mapped to single partition), fnv1a_random - FNV-1a hash of key (NULL keys are randomly partitioned).
	// Type: string
	Partitioner string `json:"partitioner"`

	// Compression codec to use for compressing message sets. inherit = inherit global compression.codec configuration.
	// range: none, gzip, snappy, lz4, zstd, inherit
	// Type: enum value
	// alias: compression.type
	CompressionCodec string `json:"compression.codec"`
}

func DefaultProducerTopicConfigHigh 

func DefaultProducerTopicConfigHigh() ProducerTopicConfigHighLevel

type ProducerTopicConfigMediumLevel 

type ProducerTopicConfigMediumLevel struct {
	// The ack timeout of the producer request in milliseconds. This value is only enforced by the broker and relies on request.required.acks being != 0.
	// range 1 ~ 9e5
	// Type: integer
	RequestTimeoutMs int `json:"request.timeout.ms"`

	// Compression level parameter for algorithm selected by configuration property compression.codec. Higher values will result in better compression at the cost of more CPU usage. Usable range is algorithm-dependent: [0-9] for gzip; [0-12] for lz4; only 0 for snappy; -1 = codec-dependent default compression level.
	// range: -1 ~ 12
	// Type: integer
	CompressionLevel int `json:"compression.level"`
}

func DefaultProducerTopicConfigMedium 

func DefaultProducerTopicConfigMedium() ProducerTopicConfigMediumLevel

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.