Kafka Logback Appender

A simple Logback appender which sends the logging output to a [Kafka] (http://kafka.apache.org) broker.

This implementation is inspired by the log4j appender included in the Kafka source distribution.
See e.g. [KafkaLog4jAppender] (https://github.com/apache/kafka/blob/trunk/log4j-appender/src/main/java/org/apache/kafka/log4jappender/KafkaLog4jAppender.java).

The Kafka Logback Appender is based on the Kafka version 2.1.0.

Configuration parameters

Appender parameters

topic

The Kafka topic the message should be published to.

SyncSend

Send message synchronously (if true, wait for completion of the returned Future).

Kafka producer parameters

The appender takes the available Kafka Producer parameters (see [ProductConfig] (https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java) class).

BrokerList   (bootstrap.servers)

List of host/port broker addresses. E.g.: host1:port1, host2:port2,....

CompressionType   (compression.type)

The compression type for all data generated by the producer. The default is none (i.e. no compression). Valid values are none, gzip, snappy, or lz4. Compression is of full batches of data, so the efficacy of batching will also impact the compression ratio (more batching means better compression).

RequiredNumAcks   (acks)

The number of acknowledgments the producer requires the leader to have received before considering a request complete. This controls the durability of records that are sent. The following settings are common:

• acks=0 If set to zero then the producer will not wait for any acknowledgment from the server at all. The record will be immediately added to the socket buffer and considered sent. No guarantee can be made that the server has received the record in this case, and the retries configuration will not take effect (as the client won\'t generally know of any failures). The offset given back for each record will always be set to -1.
• acks=1 This will mean the leader will write the record to its local log but will respond without awaiting full acknowledgement from all followers. In this case should the leader fail immediately after acknowledging the record but before the followers have replicated it then the record will be lost.
• acks=-1 This means the leader will wait for the full set of in-sync replicas to acknowledge the record. This guarantees that the record will not be lost as long as at least one in-sync replica remains alive. This is the strongest available guarantee.

maxBlockMs   (max.block.ms)

The configuration controls how long KafkaProducer.send() and KafkaProducer.partitionsFor() will block. Default: 60000 (1 minute). These methods can be blocked either because the buffer is full or metadata unavailable. Blocking in the user-supplied serializers or partitioner will not be counted against this timeout.

lingerMs   (linger.ms)

Default: 0. The producer groups together any records that arrive in between request transmissions into a single batched request. Normally this occurs only under load when records arrive faster than they can be sent out. However in some circumstances the client may want to reduce the number of requests even under moderate load. This setting accomplishes this by adding a small amount of artificial delay—that is, rather than immediately sending out a record the producer will wait for up to the given delay to allow other records to be sent so that the sends can be batched together. This can be thought of as analogous to Nagle's algorithm in TCP. This setting gives the upper bound on the delay for batching: once we get batch.size worth of records for a partition it will be sent immediately regardless of this setting, however if we have fewer than this many bytes accumulated for this partition we will 'linger' for the specified time waiting for more records to show up. This setting defaults to 0 (i.e. no delay). Setting linger.ms=5, for example, would have the effect of reducing the number of requests sent but would add up to 5ms of latency to records sent in the absence of load.

Retries   (retries)

Setting a value greater than zero will cause the client to resend any record whose send fails with a potentially transient error. Note that this retry is no different than if the client resent the record upon receiving the error. Allowing retries will potentially change the ordering of records because if two records are sent to a single partition, and the first fails and is retried but the second succeeds, then the second record may appear first.

KeySerializerClass   (key.serializer)

Serializer class for key that implements the Serializer interface. Default: org.apache.kafka.common.serialization.StringSerializer

ValueSerializerClass   (value.serializer)

Serializer class for value that implements the Serializer interface.Default: org.apache.kafka.common.serialization.ByteArraySerializer

SecurityProtocol   (security.protocol)

Protocol used to communicate with brokers. Valid values are (on this machine): PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL.

SslTruststoreLocation   (ssl.truststore.location)

The location of the trust store file.

SslTruststorePassword   (ssl.truststore.password)

The password for the trust store file.

SslKeystoreType   (ssl.keystore.type)

The file format of the key store file. This is optional for client.

SslKeystoreLocation   (ssl.keystore.location)

The location of the key store file. This is optional for client and can be used for two-way authentication for client.

SslKeystorePassword   (ssl.keystore.password)

The store password for the key store file.This is optional for client and only needed if ssl.keystore.location is configured.

SaslKerberosServiceName   (sasl.kerberos.service.name)

The Kerberos principal name that Kafka runs as. This can be defined either in Kafka's JAAS config or in Kafka's config.

Example usage

<configuration>
    <appender name="kafka-appender" class="io.github.rahulsinghai.KafkaLogbackAppender">
        <BrokerList>kafkaserver.singhaiuklimited.com:6667</BrokerList>
        <Topic>log</Topic>
        <SyncSend>false</SyncSend>

        <SecurityProtocol>SASL_PLAINTEXT</SecurityProtocol>
        <SaslKerberosServiceName>kafka</SaslKerberosServiceName>
        <ClientJaasConfPath>c:/work/jaas.conf</ClientJaasConfPath>
        <Kerb5ConfPath>c:/work/krb5.conf</Kerb5ConfPath>
        <saslMechanism>GSSAPI</saslMechanism>
        
        <encoder>
            <pattern>%date{yyyy-MM-dd'T'HH:mm:ss.SSS'Z', UTC} %-5p [%c{1}] %m</pattern>
            <charset>UTF-8</charset>
        </encoder>
    </appender>

    <root level="INFO">
        <appender-ref ref="kafka-appender"/>
    </root>
</configuration>