This repository has been archived by the owner on May 27, 2022. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 261
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
! make kafka-logback-appender generic in the event type to support al…
…ternative events like logback-access (#24). Rename PatternLayoutKafkaMessageEncoder to LayoutKafkaMessageEncoder (with deprecated class alias) (#9). Add javadoc since hint. Switch to 0.1.0-SNAPSHOT version.
- Loading branch information
1 parent
ba6f048
commit c137d62
Showing
25 changed files
with
294 additions
and
262 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
168 changes: 166 additions & 2 deletions
168
src/main/java/com/github/danielwegener/logback/kafka/KafkaAppender.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,10 +1,174 @@ | ||
| package com.github.danielwegener.logback.kafka; | ||
|
|
||
| import ch.qos.logback.classic.spi.ILoggingEvent; | ||
| import ch.qos.logback.core.Appender; | ||
| import ch.qos.logback.core.spi.AppenderAttachableImpl; | ||
| import com.github.danielwegener.logback.kafka.delivery.FailedDeliveryCallback; | ||
| import org.apache.kafka.clients.producer.KafkaProducer; | ||
| import org.apache.kafka.clients.producer.Producer; | ||
| import org.apache.kafka.clients.producer.ProducerConfig; | ||
| import org.apache.kafka.clients.producer.ProducerRecord; | ||
| import org.apache.kafka.common.KafkaException; | ||
| import org.apache.kafka.common.serialization.ByteArraySerializer; | ||
|
|
||
| import java.util.HashMap; | ||
| import java.util.Iterator; | ||
| import java.util.concurrent.ConcurrentLinkedQueue; | ||
|
|
||
| /** | ||
| * See <a href="https://github.com/danielwegener/logback-kafka-appender">logback-kafka-appender at github</a> | ||
| * @since 0.0.1 | ||
| */ | ||
| public class KafkaAppender extends KafkaAppenderBase<ILoggingEvent> { | ||
| public class KafkaAppender<E> extends KafkaAppenderConfig<E> { | ||
|
|
||
| /** | ||
| * Kafka clients uses this prefix for its slf4j logging. | ||
| * This appender defers appends of any Kafka logs since it could cause harmful infinite recursion/self feeding effects. | ||
| */ | ||
| private static final String KAFKA_LOGGER_PREFIX = "org.apache.kafka.clients"; | ||
|
|
||
| private LazyProducer lazyProducer = null; | ||
| private final AppenderAttachableImpl<E> aai = new AppenderAttachableImpl<E>(); | ||
| private final ConcurrentLinkedQueue<E> queue = new ConcurrentLinkedQueue<E>(); | ||
| private final FailedDeliveryCallback<E> failedDeliveryCallback = new FailedDeliveryCallback<E>() { | ||
| @Override | ||
| public void onFailedDelivery(E evt, Throwable throwable) { | ||
| aai.appendLoopOnAppenders(evt); | ||
| } | ||
| }; | ||
|
|
||
| public KafkaAppender() { | ||
| // setting these as config values sidesteps an unnecessary warning (minor bug in KafkaProducer) | ||
| addProducerConfigValue(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName()); | ||
| addProducerConfigValue(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName()); | ||
| } | ||
|
|
||
| @Override | ||
| public void doAppend(E e) { | ||
| ensureDeferredAppends(); | ||
| if (e instanceof ILoggingEvent && ((ILoggingEvent)e).getLoggerName().startsWith(KAFKA_LOGGER_PREFIX)) { | ||
| deferAppend(e); | ||
| } else { | ||
| super.doAppend(e); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void start() { | ||
| // only error free appenders should be activated | ||
| if (!checkPrerequisites()) return; | ||
|
|
||
| lazyProducer = new LazyProducer(); | ||
|
|
||
| super.start(); | ||
| } | ||
|
|
||
| @Override | ||
| public void stop() { | ||
| super.stop(); | ||
| if (lazyProducer != null && lazyProducer.isInitialized()) { | ||
| try { | ||
| lazyProducer.get().close(); | ||
| } catch (KafkaException e) { | ||
| this.addWarn("Failed to shut down kafka producer: " + e.getMessage(), e); | ||
| } | ||
| lazyProducer = null; | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void addAppender(Appender<E> newAppender) { | ||
| aai.addAppender(newAppender); | ||
| } | ||
|
|
||
| @Override | ||
| public Iterator<Appender<E>> iteratorForAppenders() { | ||
| return aai.iteratorForAppenders(); | ||
| } | ||
|
|
||
| @Override | ||
| public Appender<E> getAppender(String name) { | ||
| return aai.getAppender(name); | ||
| } | ||
|
|
||
| @Override | ||
| public boolean isAttached(Appender<E> appender) { | ||
| return aai.isAttached(appender); | ||
| } | ||
|
|
||
| @Override | ||
| public void detachAndStopAllAppenders() { | ||
| aai.detachAndStopAllAppenders(); | ||
| } | ||
|
|
||
| @Override | ||
| public boolean detachAppender(Appender<E> appender) { | ||
| return aai.detachAppender(appender); | ||
| } | ||
|
|
||
| @Override | ||
| public boolean detachAppender(String name) { | ||
| return aai.detachAppender(name); | ||
| } | ||
|
|
||
| @Override | ||
| protected void append(E e) { | ||
| final byte[] payload = encoder.doEncode(e); | ||
| final byte[] key = keyingStrategy.createKey(e); | ||
| final ProducerRecord<byte[], byte[]> record = new ProducerRecord<byte[],byte[]>(topic, key, payload); | ||
| deliveryStrategy.send(lazyProducer.get(), record, e, failedDeliveryCallback); | ||
| } | ||
|
|
||
| protected Producer<byte[], byte[]> createProducer() { | ||
| return new KafkaProducer<byte[], byte[]>(new HashMap<String, Object>(producerConfig)); | ||
| } | ||
|
|
||
| private void deferAppend(E event) { | ||
| queue.add(event); | ||
| } | ||
|
|
||
| // drains queue events to super | ||
| private void ensureDeferredAppends() { | ||
| E event; | ||
|
|
||
| while ((event = queue.poll()) != null) { | ||
| super.doAppend(event); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Lazy initializer for producer, patterned after commons-lang. | ||
| * | ||
| * @see <a href="https://commons.apache.org/proper/commons-lang/javadocs/api-3.4/org/apache/commons/lang3/concurrent/LazyInitializer.html">LazyInitializer</a> | ||
| */ | ||
| private class LazyProducer { | ||
|
|
||
| private volatile Producer<byte[], byte[]> producer; | ||
|
|
||
| public Producer<byte[], byte[]> get() { | ||
| Producer<byte[], byte[]> result = this.producer; | ||
| if (result == null) { | ||
| synchronized(this) { | ||
| result = this.producer; | ||
| if(result == null) { | ||
| this.producer = result = this.initialize(); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| return result; | ||
| } | ||
|
|
||
| protected Producer<byte[], byte[]> initialize() { | ||
| Producer<byte[], byte[]> producer = null; | ||
| try { | ||
| producer = createProducer(); | ||
| } catch (Exception e) { | ||
| addError("error creating producer", e); | ||
| } | ||
| return producer; | ||
| } | ||
|
|
||
| public boolean isInitialized() { return producer != null; } | ||
| } | ||
|
|
||
| } |
Oops, something went wrong.