native

README

kafkaproducer.go

Requirements

• librdkafka

Documentation

Index

• type KafkaProducer
  • func (prod *KafkaProducer) Configure(conf core.PluginConfigReader) error
  • func (prod *KafkaProducer) OnMessageDelivered(userdata []byte)
  • func (prod *KafkaProducer) OnMessageError(reason string, userdata []byte)
  • func (prod *KafkaProducer) Produce(workers *sync.WaitGroup)

Types

type KafkaProducer

type KafkaProducer struct {
	core.BufferedProducer `gollumdoc:"embed_type"`
	// contains filtered or unexported fields
}

KafkaProducer producer

NOTICE: This producer is not included in standard builds. To enable it you need to trigger a custom build with native plugins enabled. The kafka producer writes messages to a kafka cluster. This producer is backed by the native librdkafka (0.8.6) library so most settings relate to that library.

Parameters

- Servers: Defines a list of ideally all brokers in the cluster. At least one broker is required. By default this parameter is set to an empty list.

- Topic: Defines a stream to topic mapping. If a stream is not mapped the stream name is used as topic. By default this parameter is set to an empty list.

- ClientId: Sets the kafka client id used by this producer. By default this parameter is set to "gollum".

- Compression: Defines the compression algorithm to use. Possible values are "none", "zip" and "snappy". By default this parameter is set to "none".

- RequiredAcks: Defines the numbers of acknowledgements required until a message is marked as "sent". By default this parameter is set to 1.

- ServerTimeoutSec: Defines the time in seconds after which a server is defined as "not reachable". By default this parameter is set to 1.

- ServerMaxFails: Defines the number of retries after which a server is marked as "failing". By default this parameter is set to 3.

- MetadataTimeoutMs: Number of milliseconds a metadata request may take until considered as failed. By default this parameter is set to 1500.

- MetadataRefreshMs: Interval in milliseconds for querying metadata. By default this parameter is set to 300000.

- TimeoutMs: Defines the number of milliseconds to wait until a request is marked as failed. By default this parameter is set to 1500.

- Batch/TimeoutMs: Defines the number of milliseconds to wait until a batch is flushed to kafka. By default this parameter is set to 1000.

- Batch/SizeMaxKB: Defines the maximum message size in KB. Messages above this size are rejected. By default this parameter is set to 1024.

- Batch/MinMessages: Defines the minimum number of messages required for a batch to be sent. This value should be significantly lower than BatchMaxMessages to avoid messages to be rejected. By default this parameter is set to 1000.

- Batch/MaxMessages: Defines the maximum number of messages that are marked as pending at any given moment in time. If this limit is hit, additional messages will be rejected. This should be adjusted according to your maximum message throughput. By default this parameter is set to 100000.

- SendRetries: Defines the number of times librdkafka will try to re-send a message if it did not succeed. By default this parameter is set to 0.

- KeyFrom: Defines the metadata field that contains the string to be used as the key passed to kafka. When set to an empty string no key is used. By default this parameter is set to "".

- SaslMechanism: Defines the SASL mechanism to use for authentication. Accepted values are GSSAPI, PLAIN, SCRAM-SHA-256 and SCRAM-SHA-512. By default this parameter is set to "".

- SaslUsername: Sets the SASL username for use with the PLAIN mechanism. By default this parameter is set to "".

- SaslPassword: Sets the SASL password for use with the PLAIN mechanism. By default this parameter is set to "".

- SecurityProtocol: Protocol used to communicate with brokers. Accepted values are plaintext, ssl, sasl_plaintext and sasl_ssl. By default this parameter is set to "plaintext".

- SslCipherSuites: Defines the Cipher Suites to use when connection via TLS/SSL. For allowed values see man page for ciphers(1). By default this parameter is set to "".

- SslKeyLocation: Path to the client's private key (PEM) used for authentication. By default this parameter is set to "".

- SslKeyPassword: Contains the private key passphrase. By default this parameter is set to "".

- SslCertificateLocation: Path to the client's public key (PEM) used for authentication. By default this parameter is set to "".

- SslCaLocation: File or directory path to the CA certificate(s) used for verifying the broker's key. By default this parameter is set to "".

- SslCrlLocation: Path to the CRL used to verify the broker's certificate validity. By default this parameter is set to "".

Examples:

kafkaWriter:
  Type: native.KafkaProducer
  Streams: logs
  Compression: zip
  Servers:
  	- "kafka01:9092"
  	- "kafka02:9092"
  	- "kafka03:9092"
  	- "kafka04:9092"

func (*KafkaProducer) Configure

func (prod *KafkaProducer) Configure(conf core.PluginConfigReader) error

Configure initializes this producer with values from a plugin config.

func (*KafkaProducer) OnMessageDelivered

func (prod *KafkaProducer) OnMessageDelivered(userdata []byte)

OnMessageDelivered gets called by librdkafka on message delivery success

func (*KafkaProducer) OnMessageError

func (prod *KafkaProducer) OnMessageError(reason string, userdata []byte)

OnMessageError gets called by librdkafka on message delivery failure

func (*KafkaProducer) Produce

func (prod *KafkaProducer) Produce(workers *sync.WaitGroup)

Produce writes to a buffer that is sent to a given socket.