You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@logging.apache.org by pk...@apache.org on 2022/10/16 22:46:09 UTC
[logging-log4j2] branch release-2.x updated (a875db9212 -> a592301179)
This is an automated email from the ASF dual-hosted git repository.
pkarwasz pushed a change to branch release-2.x
in repository https://gitbox.apache.org/repos/asf/logging-log4j2.git
from a875db9212 Backport Kafka GitHub Actions configuration
new 7b074ed53a Replace tabs with spaces (Kafka classes)
new a592301179 [LOG4J2-2678] Backport Kafka improvements from `master`
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../mom/kafka/KafkaAppenderCloseTimeoutTest.java | 31 +-
.../core/appender/mom/kafka/KafkaAppenderTest.java | 337 ++++++++++---------
.../src/test/resources/KafkaAppenderTest.xml | 18 +-
.../core/appender/mom/kafka/KafkaAppender.java | 370 +++++++++++----------
.../core/appender/mom/kafka/KafkaManager.java | 309 +++++++++--------
src/changes/changes.xml | 5 +
6 files changed, 595 insertions(+), 475 deletions(-)
[logging-log4j2] 02/02: [LOG4J2-2678] Backport Kafka improvements from `master`
Posted by pk...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pkarwasz pushed a commit to branch release-2.x
in repository https://gitbox.apache.org/repos/asf/logging-log4j2.git
commit a5923011792bae9b4a0fd2c562a83165f319641b
Author: Piotr P. Karwasz <pi...@karwasz.org>
AuthorDate: Mon Oct 17 00:42:24 2022 +0200
[LOG4J2-2678] Backport Kafka improvements from `master`
---
.../core/appender/mom/kafka/KafkaAppenderTest.java | 1 +
.../src/test/resources/KafkaAppenderTest.xml | 6 +--
.../core/appender/mom/kafka/KafkaAppender.java | 44 ++++++++++++++----
.../core/appender/mom/kafka/KafkaManager.java | 54 ++++++++++++++++++----
src/changes/changes.xml | 5 ++
5 files changed, 88 insertions(+), 22 deletions(-)
diff --git a/log4j-core-test/src/test/java/org/apache/logging/log4j/core/appender/mom/kafka/KafkaAppenderTest.java b/log4j-core-test/src/test/java/org/apache/logging/log4j/core/appender/mom/kafka/KafkaAppenderTest.java
index 9e350d925b..b112f3028f 100644
--- a/log4j-core-test/src/test/java/org/apache/logging/log4j/core/appender/mom/kafka/KafkaAppenderTest.java
+++ b/log4j-core-test/src/test/java/org/apache/logging/log4j/core/appender/mom/kafka/KafkaAppenderTest.java
@@ -50,6 +50,7 @@ import org.junit.experimental.categories.Category;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
diff --git a/log4j-core-test/src/test/resources/KafkaAppenderTest.xml b/log4j-core-test/src/test/resources/KafkaAppenderTest.xml
index a1a9c02868..9dec545010 100644
--- a/log4j-core-test/src/test/resources/KafkaAppenderTest.xml
+++ b/log4j-core-test/src/test/resources/KafkaAppenderTest.xml
@@ -32,12 +32,12 @@
<Property name="bootstrap.servers">localhost:9092</Property>
<Property name="syncSend">false</Property>
</Kafka>
- <Kafka name="KafkaAppenderWithKey" topic="kafka-topic" key="key">
+ <Kafka name="KafkaAppenderWithKey" topic="kafka-topic" key="key" sendEventTimestamp="true">
<PatternLayout pattern="%m"/>
<Property name="timeout.ms">1000</Property>
<Property name="bootstrap.servers">localhost:9092</Property>
</Kafka>
- <Kafka name="KafkaAppenderWithKeyLookup" topic="kafka-topic" key="$${date:dd-MM-yyyy}">
+ <Kafka name="KafkaAppenderWithKeyLookup" topic="kafka-topic" key="$${date:dd-MM-yyyy}" sendEventTimestamp="true">
<PatternLayout pattern="%m"/>
<Property name="timeout.ms">1000</Property>
<Property name="bootstrap.servers">localhost:9092</Property>
@@ -47,7 +47,7 @@
<Property name="timeout.ms">1000</Property>
<Property name="bootstrap.servers">fakeLocalhost:9092</Property>
</Kafka>
- <Kafka name="KafkaAppenderNoEventTimestamp" topic="kafka-topic" key="key" eventTimestamp="false">
+ <Kafka name="KafkaAppenderNoEventTimestamp" topic="kafka-topic" key="key" sendEventTimestamp="false">
<PatternLayout pattern="%m"/>
<Property name="timeout.ms">1000</Property>
<Property name="bootstrap.servers">localhost:9092</Property>
diff --git a/log4j-core/src/main/java/org/apache/logging/log4j/core/appender/mom/kafka/KafkaAppender.java b/log4j-core/src/main/java/org/apache/logging/log4j/core/appender/mom/kafka/KafkaAppender.java
index 562dfd59e8..0253f5e5a9 100644
--- a/log4j-core/src/main/java/org/apache/logging/log4j/core/appender/mom/kafka/KafkaAppender.java
+++ b/log4j-core/src/main/java/org/apache/logging/log4j/core/appender/mom/kafka/KafkaAppender.java
@@ -37,6 +37,7 @@ import org.apache.logging.log4j.core.config.plugins.Plugin;
import org.apache.logging.log4j.core.config.plugins.PluginAttribute;
import org.apache.logging.log4j.core.config.plugins.PluginBuilderFactory;
import org.apache.logging.log4j.core.layout.SerializedLayout;
+import org.apache.logging.log4j.core.util.Integers;
/**
* Sends log events to an Apache Kafka topic.
@@ -46,14 +47,15 @@ public final class KafkaAppender extends AbstractAppender {
/**
* Builds KafkaAppender instances.
- *
- * @param <B> The type to build
+ *
+ * @param <B>
+ * The type to build
*/
public static class Builder<B extends Builder<B>> extends AbstractAppender.Builder<B>
implements org.apache.logging.log4j.core.util.Builder<KafkaAppender> {
@PluginAttribute("retryCount")
- private String retryCount;
+ private int retryCount;
@PluginAttribute("topic")
private String topic;
@@ -64,6 +66,9 @@ public final class KafkaAppender extends AbstractAppender {
@PluginAttribute(value = "syncSend", defaultBoolean = true)
private boolean syncSend;
+ @PluginAttribute(value = "sendEventTimestamp", defaultBoolean = false)
+ private boolean sendEventTimestamp;
+
@SuppressWarnings("resource")
@Override
public KafkaAppender build() {
@@ -73,7 +78,7 @@ public final class KafkaAppender extends AbstractAppender {
return null;
}
final KafkaManager kafkaManager = KafkaManager.getManager(getConfiguration().getLoggerContext(), getName(),
- topic, syncSend, getPropertyArray(), key);
+ topic, syncSend, sendEventTimestamp, getPropertyArray(), key);
return new KafkaAppender(getName(), layout, getFilter(), isIgnoreExceptions(), kafkaManager,
getPropertyArray(), getRetryCount());
}
@@ -86,13 +91,16 @@ public final class KafkaAppender extends AbstractAppender {
}
return intRetryCount;
-
}
public String getTopic() {
return topic;
}
+ public boolean isSendEventTimestamp() {
+ return sendEventTimestamp;
+ }
+
public boolean isSyncSend() {
return syncSend;
}
@@ -102,6 +110,22 @@ public final class KafkaAppender extends AbstractAppender {
return asBuilder();
}
+ @Deprecated
+ public B setRetryCount(final String retryCount) {
+ this.retryCount = Integers.parseInt(retryCount, 0);
+ return asBuilder();
+ }
+
+ public B setRetryCount(final int retryCount) {
+ this.retryCount = retryCount;
+ return asBuilder();
+ }
+
+ public B setSendEventTimestamp(boolean sendEventTimestamp) {
+ this.sendEventTimestamp = sendEventTimestamp;
+ return asBuilder();
+ }
+
public B setSyncSend(final boolean syncSend) {
this.syncSend = syncSend;
return asBuilder();
@@ -128,7 +152,7 @@ public final class KafkaAppender extends AbstractAppender {
}
final KafkaManager kafkaManager = KafkaManager.getManager(configuration.getLoggerContext(), name, topic, true,
properties, key);
- return new KafkaAppender(name, layout, filter, ignoreExceptions, kafkaManager, null, null);
+ return new KafkaAppender(name, layout, filter, ignoreExceptions, kafkaManager, null, 0);
}
/**
@@ -143,7 +167,7 @@ public final class KafkaAppender extends AbstractAppender {
/**
* Creates a builder for a KafkaAppender.
- *
+ *
* @return a builder for a KafkaAppender.
*/
@PluginBuilderFactory
@@ -151,13 +175,13 @@ public final class KafkaAppender extends AbstractAppender {
return new Builder<B>().asBuilder();
}
- private final Integer retryCount;
+ private final Integer retryCount;
private final KafkaManager manager;
private KafkaAppender(final String name, final Layout<? extends Serializable> layout, final Filter filter,
final boolean ignoreExceptions, final KafkaManager manager, final Property[] properties,
- final Integer retryCount) {
+ final int retryCount) {
super(name, filter, layout, ignoreExceptions, properties);
this.manager = Objects.requireNonNull(manager, "manager");
this.retryCount = retryCount;
@@ -221,6 +245,6 @@ public final class KafkaAppender extends AbstractAppender {
} else {
data = layout.toByteArray(event);
}
- manager.send(data);
+ manager.send(data, event.getTimeMillis());
}
}
diff --git a/log4j-core/src/main/java/org/apache/logging/log4j/core/appender/mom/kafka/KafkaManager.java b/log4j-core/src/main/java/org/apache/logging/log4j/core/appender/mom/kafka/KafkaManager.java
index 378099e734..ab814e460e 100644
--- a/log4j-core/src/main/java/org/apache/logging/log4j/core/appender/mom/kafka/KafkaManager.java
+++ b/log4j-core/src/main/java/org/apache/logging/log4j/core/appender/mom/kafka/KafkaManager.java
@@ -25,8 +25,10 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
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.clients.producer.RecordMetadata;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.logging.log4j.core.LoggerContext;
import org.apache.logging.log4j.core.appender.AbstractManager;
import org.apache.logging.log4j.core.appender.ManagerFactory;
@@ -50,6 +52,8 @@ public class KafkaManager extends AbstractManager {
private final String topic;
private final String key;
private final boolean syncSend;
+ private final boolean sendTimestamp;
+
private static final KafkaManagerFactory factory = new KafkaManagerFactory();
/*
@@ -58,13 +62,19 @@ public class KafkaManager extends AbstractManager {
*/
public KafkaManager(final LoggerContext loggerContext, final String name, final String topic,
final boolean syncSend, final Property[] properties, final String key) {
+ this(loggerContext, name, topic, syncSend, false, properties, key);
+ }
+
+ private KafkaManager(final LoggerContext loggerContext, final String name, final String topic, final boolean syncSend,
+ final boolean sendTimestamp, final Property[] properties, final String key) {
super(loggerContext, name);
this.topic = Objects.requireNonNull(topic, "topic");
this.syncSend = syncSend;
+ this.sendTimestamp = sendTimestamp;
- config.setProperty("key.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
- config.setProperty("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
- config.setProperty("batch.size", "0");
+ config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
+ config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
+ config.put(ProducerConfig.BATCH_SIZE_CONFIG, 0);
for (final Property property : properties) {
config.setProperty(property.getName(), property.getValue());
@@ -72,7 +82,11 @@ public class KafkaManager extends AbstractManager {
this.key = key;
- this.timeoutMillis = Integers.parseInt(config.getProperty("timeout.ms", DEFAULT_TIMEOUT_MILLIS));
+ String timeoutMillis = config.getProperty("timeout.ms");
+ if (timeoutMillis == null) {
+ timeoutMillis = config.getProperty(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, DEFAULT_TIMEOUT_MILLIS);
+ }
+ this.timeoutMillis = Integers.parseInt(timeoutMillis);
}
@Override
@@ -105,7 +119,12 @@ public class KafkaManager extends AbstractManager {
}
}
+ @Deprecated
public void send(final byte[] msg) throws ExecutionException, InterruptedException, TimeoutException {
+ send(msg, null);
+ }
+
+ public void send(final byte[] msg, final Long eventTimestamp) throws ExecutionException, InterruptedException, TimeoutException {
if (producer != null) {
byte[] newKey = null;
@@ -116,7 +135,9 @@ public class KafkaManager extends AbstractManager {
newKey = key.getBytes(StandardCharsets.UTF_8);
}
- final ProducerRecord<byte[], byte[]> newRecord = new ProducerRecord<>(topic, newKey, msg);
+ final Long timestamp = sendTimestamp ? eventTimestamp : null;
+
+ final ProducerRecord<byte[], byte[]> newRecord = new ProducerRecord<>(topic, null, timestamp, newKey, msg);
if (syncSend) {
final Future<RecordMetadata> response = producer.send(newRecord);
response.get(timeoutMillis, TimeUnit.MILLISECONDS);
@@ -140,28 +161,42 @@ public class KafkaManager extends AbstractManager {
return topic;
}
+ @Deprecated
public static KafkaManager getManager(final LoggerContext loggerContext, final String name, final String topic,
final boolean syncSend, final Property[] properties, final String key) {
+ return getManager(loggerContext, name, topic, syncSend, false, properties, key);
+ }
+
+ static KafkaManager getManager(final LoggerContext loggerContext, final String name, final String topic,
+ final boolean syncSend, final boolean sendTimestamp, final Property[] properties, final String key) {
StringBuilder sb = new StringBuilder(name);
- sb.append(" ").append(topic).append(" ").append(syncSend + "");
+ sb.append(" ")
+ .append(topic)
+ .append(" ")
+ .append(syncSend)
+ .append(" ")
+ .append(sendTimestamp);
for (Property prop : properties) {
sb.append(" ").append(prop.getName()).append("=").append(prop.getValue());
}
- return getManager(sb.toString(), factory, new FactoryData(loggerContext, topic, syncSend, properties, key));
+ return getManager(sb.toString(), factory, new FactoryData(loggerContext, topic, syncSend, sendTimestamp,
+ properties, key));
}
private static class FactoryData {
private final LoggerContext loggerContext;
private final String topic;
private final boolean syncSend;
+ private final boolean sendTimestamp;
private final Property[] properties;
private final String key;
public FactoryData(final LoggerContext loggerContext, final String topic, final boolean syncSend,
- final Property[] properties, final String key) {
+ final boolean sendTimestamp, final Property[] properties, final String key) {
this.loggerContext = loggerContext;
this.topic = topic;
this.syncSend = syncSend;
+ this.sendTimestamp = sendTimestamp;
this.properties = properties;
this.key = key;
}
@@ -171,7 +206,8 @@ public class KafkaManager extends AbstractManager {
private static class KafkaManagerFactory implements ManagerFactory<KafkaManager, FactoryData> {
@Override
public KafkaManager createManager(String name, FactoryData data) {
- return new KafkaManager(data.loggerContext, name, data.topic, data.syncSend, data.properties, data.key);
+ return new KafkaManager(data.loggerContext, name, data.topic, data.syncSend, data.sendTimestamp,
+ data.properties, data.key);
}
}
diff --git a/src/changes/changes.xml b/src/changes/changes.xml
index 49c2090af1..cd68354d80 100644
--- a/src/changes/changes.xml
+++ b/src/changes/changes.xml
@@ -29,6 +29,11 @@
- "update" - Change
- "remove" - Removed
-->
+ <release version="2.19.1" date="TBD" description="GA Release 2.19.1">
+ <action issue="LOG4J2-2678" dev="pkarwasz" type="update" due-to="Federico D'Ambrosio">
+ Add LogEvent timestamp to ProducerRecord in KafkaAppender.
+ </action>
+ </release>
<release version="2.19.0" date="2022-09-09" description="GA Release 2.19.0">
<action issue="LOG4J2-3614" dev="vy" type="fix" due-to="strainu">
Harden InstantFormatter against delegate failures.
[logging-log4j2] 01/02: Replace tabs with spaces (Kafka classes)
Posted by pk...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pkarwasz pushed a commit to branch release-2.x
in repository https://gitbox.apache.org/repos/asf/logging-log4j2.git
commit 7b074ed53aa9993e52e9ebe279fcd39b3f5a8dbb
Author: Piotr P. Karwasz <pi...@karwasz.org>
AuthorDate: Sun Oct 16 12:26:24 2022 +0200
Replace tabs with spaces (Kafka classes)
---
.../mom/kafka/KafkaAppenderCloseTimeoutTest.java | 31 +-
.../core/appender/mom/kafka/KafkaAppenderTest.java | 336 +++++++++++---------
.../src/test/resources/KafkaAppenderTest.xml | 14 +-
.../core/appender/mom/kafka/KafkaAppender.java | 352 ++++++++++-----------
.../core/appender/mom/kafka/KafkaManager.java | 273 ++++++++--------
5 files changed, 530 insertions(+), 476 deletions(-)
diff --git a/log4j-core-test/src/test/java/org/apache/logging/log4j/core/appender/mom/kafka/KafkaAppenderCloseTimeoutTest.java b/log4j-core-test/src/test/java/org/apache/logging/log4j/core/appender/mom/kafka/KafkaAppenderCloseTimeoutTest.java
index e391fe34f5..9f2baf5520 100644
--- a/log4j-core-test/src/test/java/org/apache/logging/log4j/core/appender/mom/kafka/KafkaAppenderCloseTimeoutTest.java
+++ b/log4j-core-test/src/test/java/org/apache/logging/log4j/core/appender/mom/kafka/KafkaAppenderCloseTimeoutTest.java
@@ -17,9 +17,14 @@
package org.apache.logging.log4j.core.appender.mom.kafka;
+import java.time.Duration;
+import java.util.Properties;
import java.util.concurrent.TimeUnit;
import org.apache.kafka.clients.producer.MockProducer;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.apache.kafka.common.serialization.Serializer;
import org.apache.logging.log4j.core.Appender;
import org.apache.logging.log4j.core.test.categories.Appenders;
import org.apache.logging.log4j.core.test.junit.LoggerContextRule;
@@ -32,20 +37,34 @@ import org.junit.experimental.categories.Category;
@Category(Appenders.Kafka.class)
public class KafkaAppenderCloseTimeoutTest {
- private static final MockProducer<byte[], byte[]> kafka = new MockProducer<byte[], byte[]>(true, null, null) {
+ private static final Serializer<byte[]> SERIALIZER = new ByteArraySerializer();
+
+ private static final MockProducer<byte[], byte[]> kafka = new MockProducer<byte[], byte[]>(true, SERIALIZER,
+ SERIALIZER) {
@Override
public void close() {
try {
Thread.sleep(3000);
- } catch (final InterruptedException ignore) {
+ } catch (InterruptedException ignore) {
+ // NOP
}
}
- @Override
- public void close(final long timeout, final TimeUnit timeUnit) {
+ // @Override in version 3.3.1
+ public void close(Duration timeout) {
+ try {
+ Thread.sleep(timeout.toMillis());
+ } catch (InterruptedException ignore) {
+ // NOP
+ }
+ }
+
+ // @Override in version 1.1.1
+ public void close(long timeout, TimeUnit timeUnit) {
try {
Thread.sleep(timeUnit.toMillis(timeout));
- } catch (final InterruptedException ignore) {
+ } catch (InterruptedException ignore) {
+ // NOP
}
}
};
@@ -68,4 +87,4 @@ public class KafkaAppenderCloseTimeoutTest {
final Appender appender = ctx.getRequiredAppender("KafkaAppender");
appender.stop();
}
-}
\ No newline at end of file
+}
diff --git a/log4j-core-test/src/test/java/org/apache/logging/log4j/core/appender/mom/kafka/KafkaAppenderTest.java b/log4j-core-test/src/test/java/org/apache/logging/log4j/core/appender/mom/kafka/KafkaAppenderTest.java
index 26216db391..9e350d925b 100644
--- a/log4j-core-test/src/test/java/org/apache/logging/log4j/core/appender/mom/kafka/KafkaAppenderTest.java
+++ b/log4j-core-test/src/test/java/org/apache/logging/log4j/core/appender/mom/kafka/KafkaAppenderTest.java
@@ -14,19 +14,14 @@
* See the license for the specific language governing permissions and
* limitations under the license.
*/
-
package org.apache.logging.log4j.core.appender.mom.kafka;
-import static org.junit.Assert.assertArrayEquals;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
-
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.ObjectInput;
import java.nio.charset.StandardCharsets;
import java.text.SimpleDateFormat;
+import java.time.Duration;
import java.util.Date;
import java.util.List;
import java.util.concurrent.Future;
@@ -36,6 +31,8 @@ import java.util.concurrent.TimeoutException;
import org.apache.kafka.clients.producer.MockProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.apache.kafka.common.serialization.Serializer;
import org.apache.logging.log4j.Level;
import org.apache.logging.log4j.ThreadContext;
import org.apache.logging.log4j.core.Appender;
@@ -51,155 +48,188 @@ import org.junit.Rule;
import org.junit.Test;
import org.junit.experimental.categories.Category;
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+
@Category(Appenders.Kafka.class)
public class KafkaAppenderTest {
- private static final MockProducer<byte[], byte[]> kafka = new MockProducer<byte[], byte[]>(true, null, null) {
-
- @Override
- public synchronized Future<RecordMetadata> send(ProducerRecord<byte[], byte[]> record) {
-
- Future<RecordMetadata> retVal = super.send(record);
-
- boolean isRetryTest = "true".equals(ThreadContext.get("KafkaAppenderWithRetryCount"));
- if (isRetryTest) {
- try {
- throw new TimeoutException();
- } catch (TimeoutException e) {
- // TODO Auto-generated catch block
- throw new RuntimeException(e);
- }
- }
-
- return retVal;
- };
-
- @Override
- public void close(final long timeout, final TimeUnit timeUnit) {
- }
-
- };
-
- private static final String LOG_MESSAGE = "Hello, world!";
- private static final String TOPIC_NAME = "kafka-topic";
- private static final int RETRY_COUNT = 3;
-
- private static Log4jLogEvent createLogEvent() {
- return Log4jLogEvent.newBuilder().setLoggerName(KafkaAppenderTest.class.getName())
- .setLoggerFqcn(KafkaAppenderTest.class.getName()).setLevel(Level.INFO)
- .setMessage(new SimpleMessage(LOG_MESSAGE)).build();
- }
-
- @BeforeClass
- public static void setUpClass() throws Exception {
- KafkaManager.producerFactory = config -> kafka;
- }
-
- @Rule
- public LoggerContextRule ctx = new LoggerContextRule("KafkaAppenderTest.xml");
-
- @Before
- public void setUp() throws Exception {
- kafka.clear();
- }
-
- @Test
- public void testAppendWithLayout() throws Exception {
- final Appender appender = ctx.getRequiredAppender("KafkaAppenderWithLayout");
- appender.append(createLogEvent());
- final List<ProducerRecord<byte[], byte[]>> history = kafka.history();
- assertEquals(1, history.size());
- final ProducerRecord<byte[], byte[]> item = history.get(0);
- assertNotNull(item);
- assertEquals(TOPIC_NAME, item.topic());
- assertNull(item.key());
- assertEquals("[" + LOG_MESSAGE + "]", new String(item.value(), StandardCharsets.UTF_8));
- }
-
- @Test
- public void testAppendWithSerializedLayout() throws Exception {
- final Appender appender = ctx.getRequiredAppender("KafkaAppenderWithSerializedLayout");
- final LogEvent logEvent = createLogEvent();
- appender.append(logEvent);
- final List<ProducerRecord<byte[], byte[]>> history = kafka.history();
- assertEquals(1, history.size());
- final ProducerRecord<byte[], byte[]> item = history.get(0);
- assertNotNull(item);
- assertEquals(TOPIC_NAME, item.topic());
- assertNull(item.key());
- assertEquals(LOG_MESSAGE, deserializeLogEvent(item.value()).getMessage().getFormattedMessage());
- }
-
- @Test
- public void testAsyncAppend() throws Exception {
- final Appender appender = ctx.getRequiredAppender("AsyncKafkaAppender");
- appender.append(createLogEvent());
- final List<ProducerRecord<byte[], byte[]>> history = kafka.history();
- assertEquals(1, history.size());
- final ProducerRecord<byte[], byte[]> item = history.get(0);
- assertNotNull(item);
- assertEquals(TOPIC_NAME, item.topic());
- assertNull(item.key());
- assertEquals(LOG_MESSAGE, new String(item.value(), StandardCharsets.UTF_8));
- }
-
- @Test
- public void testAppendWithKey() throws Exception {
- final Appender appender = ctx.getRequiredAppender("KafkaAppenderWithKey");
- final LogEvent logEvent = createLogEvent();
- appender.append(logEvent);
- final List<ProducerRecord<byte[], byte[]>> history = kafka.history();
- assertEquals(1, history.size());
- final ProducerRecord<byte[], byte[]> item = history.get(0);
- assertNotNull(item);
- assertEquals(TOPIC_NAME, item.topic());
- final String msgKey = item.key().toString();
- final byte[] keyValue = "key".getBytes(StandardCharsets.UTF_8);
- assertArrayEquals(item.key(), keyValue);
- assertEquals(LOG_MESSAGE, new String(item.value(), StandardCharsets.UTF_8));
- }
-
- @Test
- public void testAppendWithKeyLookup() throws Exception {
- final Appender appender = ctx.getRequiredAppender("KafkaAppenderWithKeyLookup");
- final LogEvent logEvent = createLogEvent();
- final Date date = new Date();
- final SimpleDateFormat format = new SimpleDateFormat("dd-MM-yyyy");
- appender.append(logEvent);
- final List<ProducerRecord<byte[], byte[]>> history = kafka.history();
- assertEquals(1, history.size());
- final ProducerRecord<byte[], byte[]> item = history.get(0);
- assertNotNull(item);
- assertEquals(TOPIC_NAME, item.topic());
- final byte[] keyValue = format.format(date).getBytes(StandardCharsets.UTF_8);
- assertArrayEquals(item.key(), keyValue);
- assertEquals(LOG_MESSAGE, new String(item.value(), StandardCharsets.UTF_8));
- }
-
- @Test
- public void testAppendWithRetryCount() {
- try {
- ThreadContext.put("KafkaAppenderWithRetryCount", "true");
- final Appender appender = ctx.getRequiredAppender("KafkaAppenderWithRetryCount");
- final LogEvent logEvent = createLogEvent();
- appender.append(logEvent);
-
- final List<ProducerRecord<byte[], byte[]>> history = kafka.history();
- assertEquals(RETRY_COUNT + 1, history.size());
- } catch (Exception e) {
- e.printStackTrace();
- } finally {
- ThreadContext.clearMap();
- }
-
- }
-
- private LogEvent deserializeLogEvent(final byte[] data) throws IOException, ClassNotFoundException {
- final ByteArrayInputStream bis = new ByteArrayInputStream(data);
- try (ObjectInput ois = new FilteredObjectInputStream(bis)) {
- return (LogEvent) ois.readObject();
- }
- }
+ private static final Serializer<byte[]> SERIALIZER = new ByteArraySerializer();
+
+ private static final MockProducer<byte[], byte[]> kafka = new MockProducer<byte[], byte[]>(true, SERIALIZER,
+ SERIALIZER) {
+
+ @Override
+ public synchronized Future<RecordMetadata> send(ProducerRecord<byte[], byte[]> record) {
+
+ Future<RecordMetadata> retVal = super.send(record);
+
+ boolean isRetryTest = "true".equals(ThreadContext.get("KafkaAppenderWithRetryCount"));
+ if (isRetryTest) {
+ try {
+ throw new TimeoutException();
+ } catch (TimeoutException e) {
+ // TODO Auto-generated catch block
+ throw new RuntimeException(e);
+ }
+ }
+
+ return retVal;
+ };
+
+ // @Override in version 1.1.1
+ public void close(final long timeout, final TimeUnit timeUnit) {
+ // Intentionally do not close in order to reuse
+ }
+
+ // @Override in version 3.3.1
+ public void close(final Duration timeout) {
+ // Intentionally do no close in order to reuse
+ }
+ };
+
+ private static final String LOG_MESSAGE = "Hello, world!";
+ private static final String TOPIC_NAME = "kafka-topic";
+ private static final int RETRY_COUNT = 3;
+
+ private static Log4jLogEvent createLogEvent() {
+ return Log4jLogEvent.newBuilder()
+ .setLoggerName(KafkaAppenderTest.class.getName())
+ .setLoggerFqcn(KafkaAppenderTest.class.getName())
+ .setLevel(Level.INFO)
+ .setMessage(new SimpleMessage(LOG_MESSAGE))
+ .build();
+ }
+
+ @BeforeClass
+ public static void setUpClass() throws Exception {
+ KafkaManager.producerFactory = config -> kafka;
+ }
+
+ @Rule
+ public LoggerContextRule ctx = new LoggerContextRule("KafkaAppenderTest.xml");
+
+ @Before
+ public void setUp() throws Exception {
+ kafka.clear();
+ }
+
+ @Test
+ public void testAppendWithLayout() throws Exception {
+ final Appender appender = ctx.getRequiredAppender("KafkaAppenderWithLayout");
+ appender.append(createLogEvent());
+ final List<ProducerRecord<byte[], byte[]>> history = kafka.history();
+ assertEquals(1, history.size());
+ final ProducerRecord<byte[], byte[]> item = history.get(0);
+ assertNotNull(item);
+ assertEquals(TOPIC_NAME, item.topic());
+ assertNull(item.key());
+ assertEquals("[" + LOG_MESSAGE + "]", new String(item.value(), StandardCharsets.UTF_8));
+ }
+
+ @Test
+ public void testAppendWithSerializedLayout() throws Exception {
+ final Appender appender = ctx.getRequiredAppender("KafkaAppenderWithSerializedLayout");
+ final LogEvent logEvent = createLogEvent();
+ appender.append(logEvent);
+ final List<ProducerRecord<byte[], byte[]>> history = kafka.history();
+ assertEquals(1, history.size());
+ final ProducerRecord<byte[], byte[]> item = history.get(0);
+ assertNotNull(item);
+ assertEquals(TOPIC_NAME, item.topic());
+ assertNull(item.key());
+ assertEquals(LOG_MESSAGE, deserializeLogEvent(item.value()).getMessage().getFormattedMessage());
+ }
+
+ @Test
+ public void testAsyncAppend() throws Exception {
+ final Appender appender = ctx.getRequiredAppender("AsyncKafkaAppender");
+ appender.append(createLogEvent());
+ final List<ProducerRecord<byte[], byte[]>> history = kafka.history();
+ assertEquals(1, history.size());
+ final ProducerRecord<byte[], byte[]> item = history.get(0);
+ assertNotNull(item);
+ assertEquals(TOPIC_NAME, item.topic());
+ assertNull(item.key());
+ assertEquals(LOG_MESSAGE, new String(item.value(), StandardCharsets.UTF_8));
+ }
+
+ @Test
+ public void testAppendWithKey() throws Exception {
+ final Appender appender = ctx.getRequiredAppender("KafkaAppenderWithKey");
+ final LogEvent logEvent = createLogEvent();
+ appender.append(logEvent);
+ final List<ProducerRecord<byte[], byte[]>> history = kafka.history();
+ assertEquals(1, history.size());
+ final ProducerRecord<byte[], byte[]> item = history.get(0);
+ assertNotNull(item);
+ assertEquals(TOPIC_NAME, item.topic());
+ final byte[] keyValue = "key".getBytes(StandardCharsets.UTF_8);
+ assertEquals(Long.valueOf(logEvent.getTimeMillis()), item.timestamp());
+ assertArrayEquals(item.key(), keyValue);
+ assertEquals(LOG_MESSAGE, new String(item.value(), StandardCharsets.UTF_8));
+ }
+
+ @Test
+ public void testAppendWithKeyLookup() throws Exception {
+ final Appender appender = ctx.getRequiredAppender("KafkaAppenderWithKeyLookup");
+ final LogEvent logEvent = createLogEvent();
+ final Date date = new Date();
+ final SimpleDateFormat format = new SimpleDateFormat("dd-MM-yyyy");
+ appender.append(logEvent);
+ final List<ProducerRecord<byte[], byte[]>> history = kafka.history();
+ assertEquals(1, history.size());
+ final ProducerRecord<byte[], byte[]> item = history.get(0);
+ assertNotNull(item);
+ assertEquals(TOPIC_NAME, item.topic());
+ final byte[] keyValue = format.format(date).getBytes(StandardCharsets.UTF_8);
+ assertEquals(Long.valueOf(logEvent.getTimeMillis()), item.timestamp());
+ assertArrayEquals(item.key(), keyValue);
+ assertEquals(LOG_MESSAGE, new String(item.value(), StandardCharsets.UTF_8));
+ }
+
+ @Test
+ public void testAppendWithRetryCount() {
+ try {
+ ThreadContext.put("KafkaAppenderWithRetryCount", "true");
+ final Appender appender = ctx.getRequiredAppender("KafkaAppenderWithRetryCount");
+ final LogEvent logEvent = createLogEvent();
+ appender.append(logEvent);
+
+ final List<ProducerRecord<byte[], byte[]>> history = kafka.history();
+ assertEquals(RETRY_COUNT + 1, history.size());
+ } catch (Exception e) {
+ e.printStackTrace();
+ } finally {
+ ThreadContext.clearMap();
+ }
+
+ }
+
+ @Test
+ public void testAppenderNoEventTimestamp() throws Exception {
+ final Appender appender = ctx.getRequiredAppender("KafkaAppenderNoEventTimestamp");
+ final LogEvent logEvent = createLogEvent();
+ appender.append(logEvent);
+ final List<ProducerRecord<byte[], byte[]>> history = kafka.history();
+ assertEquals(1, history.size());
+ final ProducerRecord<byte[], byte[]> item = history.get(0);
+ assertNotNull(item);
+ assertEquals(TOPIC_NAME, item.topic());
+ byte[] keyValue = "key".getBytes(StandardCharsets.UTF_8);
+ assertArrayEquals(item.key(), keyValue);
+ assertNotEquals(Long.valueOf(logEvent.getTimeMillis()), item.timestamp());
+ assertEquals(LOG_MESSAGE, new String(item.value(), StandardCharsets.UTF_8));
+ }
+
+ private LogEvent deserializeLogEvent(final byte[] data) throws IOException, ClassNotFoundException {
+ final ByteArrayInputStream bis = new ByteArrayInputStream(data);
+ try (ObjectInput ois = new FilteredObjectInputStream(bis)) {
+ return (LogEvent) ois.readObject();
+ }
+ }
// public void shouldRetryWhenTimeoutExceptionOccursOnSend() throws Exception {
// final AtomicInteger attempt = new AtomicInteger(0);
@@ -220,4 +250,4 @@ public class KafkaAppenderTest {
// assertEquals(Long.valueOf(0L), offset);
// }
-}
\ No newline at end of file
+}
diff --git a/log4j-core-test/src/test/resources/KafkaAppenderTest.xml b/log4j-core-test/src/test/resources/KafkaAppenderTest.xml
index 7b3af772db..a1a9c02868 100644
--- a/log4j-core-test/src/test/resources/KafkaAppenderTest.xml
+++ b/log4j-core-test/src/test/resources/KafkaAppenderTest.xml
@@ -17,7 +17,7 @@
-->
<Configuration name="KafkaAppenderTest" status="OFF">
<Appenders>
- <Kafka name="KafkaAppenderWithLayout" topic="kafka-topic">
+ <Kafka name="KafkaAppenderWithLayout" topic="kafka-topic">
<PatternLayout pattern="[%m]"/>
<Property name="bootstrap.servers">localhost:9092</Property>
<Property name="timeout.ms">1000</Property>
@@ -42,10 +42,15 @@
<Property name="timeout.ms">1000</Property>
<Property name="bootstrap.servers">localhost:9092</Property>
</Kafka>
- <Kafka name="KafkaAppenderWithRetryCount" topic="kafka-topic" retryCount="3" >
+ <Kafka name="KafkaAppenderWithRetryCount" topic="kafka-topic" retryCount="3" >
<PatternLayout pattern="%m"/>
<Property name="timeout.ms">1000</Property>
- <Property name="bootstrap.servers">fakeLoclhost:9092</Property>
+ <Property name="bootstrap.servers">fakeLocalhost:9092</Property>
+ </Kafka>
+ <Kafka name="KafkaAppenderNoEventTimestamp" topic="kafka-topic" key="key" eventTimestamp="false">
+ <PatternLayout pattern="%m"/>
+ <Property name="timeout.ms">1000</Property>
+ <Property name="bootstrap.servers">localhost:9092</Property>
</Kafka>
</Appenders>
<Loggers>
@@ -54,7 +59,8 @@
<AppenderRef ref="KafkaAppenderWithSerializedLayout"/>
<AppenderRef ref="AsyncKafkaAppender"/>
<AppenderRef ref="KafkaAppenderWithKey"/>
- <AppenderRef ref="KafkaAppenderWithRetryCount"/>
+ <AppenderRef ref="KafkaAppenderWithRetryCount"/>
+ <AppenderRef ref="KafkaAppenderNoEventTimestamp"/>
</Root>
</Loggers>
</Configuration>
diff --git a/log4j-core/src/main/java/org/apache/logging/log4j/core/appender/mom/kafka/KafkaAppender.java b/log4j-core/src/main/java/org/apache/logging/log4j/core/appender/mom/kafka/KafkaAppender.java
index d925d5cb4d..562dfd59e8 100644
--- a/log4j-core/src/main/java/org/apache/logging/log4j/core/appender/mom/kafka/KafkaAppender.java
+++ b/log4j-core/src/main/java/org/apache/logging/log4j/core/appender/mom/kafka/KafkaAppender.java
@@ -45,182 +45,182 @@ import org.apache.logging.log4j.core.layout.SerializedLayout;
public final class KafkaAppender extends AbstractAppender {
/**
- * Builds KafkaAppender instances.
- *
- * @param <B> The type to build
- */
- public static class Builder<B extends Builder<B>> extends AbstractAppender.Builder<B>
- implements org.apache.logging.log4j.core.util.Builder<KafkaAppender> {
-
- @PluginAttribute("retryCount")
- private String retryCount;
-
- @PluginAttribute("topic")
- private String topic;
-
- @PluginAttribute("key")
- private String key;
-
- @PluginAttribute(value = "syncSend", defaultBoolean = true)
- private boolean syncSend;
-
- @SuppressWarnings("resource")
- @Override
- public KafkaAppender build() {
- final Layout<? extends Serializable> layout = getLayout();
- if (layout == null) {
- AbstractLifeCycle.LOGGER.error("No layout provided for KafkaAppender");
- return null;
- }
- final KafkaManager kafkaManager = KafkaManager.getManager(getConfiguration().getLoggerContext(), getName(),
- topic, syncSend, getPropertyArray(), key);
- return new KafkaAppender(getName(), layout, getFilter(), isIgnoreExceptions(), kafkaManager,
- getPropertyArray(), getRetryCount());
- }
-
- public Integer getRetryCount() {
- Integer intRetryCount = null;
- try {
- intRetryCount = Integer.valueOf(retryCount);
- } catch (NumberFormatException e) {
-
- }
- return intRetryCount;
-
- }
-
- public String getTopic() {
- return topic;
- }
-
- public boolean isSyncSend() {
- return syncSend;
- }
-
- public B setKey(final String key) {
- this.key = key;
- return asBuilder();
- }
-
- public B setSyncSend(final boolean syncSend) {
- this.syncSend = syncSend;
- return asBuilder();
- }
-
- public B setTopic(final String topic) {
- this.topic = topic;
- return asBuilder();
- }
-
- }
-
- private static final String[] KAFKA_CLIENT_PACKAGES = new String[] { "org.apache.kafka.common", "org.apache.kafka.clients" };
-
- @Deprecated
- public static KafkaAppender createAppender(final Layout<? extends Serializable> layout, final Filter filter,
- final String name, final boolean ignoreExceptions, final String topic, final Property[] properties,
- final Configuration configuration,
- final String key) {
-
- if (layout == null) {
- AbstractLifeCycle.LOGGER.error("No layout provided for KafkaAppender");
- return null;
- }
- final KafkaManager kafkaManager = KafkaManager.getManager(configuration.getLoggerContext(), name, topic, true,
- properties, key);
- return new KafkaAppender(name, layout, filter, ignoreExceptions, kafkaManager, null, null);
- }
-
- /**
- * Tests if the given log event is from a Kafka Producer implementation.
- *
- * @param event The event to test.
- * @return true to avoid recursion and skip logging, false to log.
- */
- private static boolean isRecursive(final LogEvent event) {
- return Stream.of(KAFKA_CLIENT_PACKAGES).anyMatch(prefix -> event.getLoggerName().startsWith(prefix));
+ * Builds KafkaAppender instances.
+ *
+ * @param <B> The type to build
+ */
+ public static class Builder<B extends Builder<B>> extends AbstractAppender.Builder<B>
+ implements org.apache.logging.log4j.core.util.Builder<KafkaAppender> {
+
+ @PluginAttribute("retryCount")
+ private String retryCount;
+
+ @PluginAttribute("topic")
+ private String topic;
+
+ @PluginAttribute("key")
+ private String key;
+
+ @PluginAttribute(value = "syncSend", defaultBoolean = true)
+ private boolean syncSend;
+
+ @SuppressWarnings("resource")
+ @Override
+ public KafkaAppender build() {
+ final Layout<? extends Serializable> layout = getLayout();
+ if (layout == null) {
+ AbstractLifeCycle.LOGGER.error("No layout provided for KafkaAppender");
+ return null;
+ }
+ final KafkaManager kafkaManager = KafkaManager.getManager(getConfiguration().getLoggerContext(), getName(),
+ topic, syncSend, getPropertyArray(), key);
+ return new KafkaAppender(getName(), layout, getFilter(), isIgnoreExceptions(), kafkaManager,
+ getPropertyArray(), getRetryCount());
+ }
+
+ public Integer getRetryCount() {
+ Integer intRetryCount = null;
+ try {
+ intRetryCount = Integer.valueOf(retryCount);
+ } catch (NumberFormatException e) {
+
+ }
+ return intRetryCount;
+
+ }
+
+ public String getTopic() {
+ return topic;
+ }
+
+ public boolean isSyncSend() {
+ return syncSend;
+ }
+
+ public B setKey(final String key) {
+ this.key = key;
+ return asBuilder();
+ }
+
+ public B setSyncSend(final boolean syncSend) {
+ this.syncSend = syncSend;
+ return asBuilder();
+ }
+
+ public B setTopic(final String topic) {
+ this.topic = topic;
+ return asBuilder();
+ }
+
+ }
+
+ private static final String[] KAFKA_CLIENT_PACKAGES = new String[] { "org.apache.kafka.common",
+ "org.apache.kafka.clients" };
+
+ @Deprecated
+ public static KafkaAppender createAppender(final Layout<? extends Serializable> layout, final Filter filter,
+ final String name, final boolean ignoreExceptions, final String topic, final Property[] properties,
+ final Configuration configuration, final String key) {
+
+ if (layout == null) {
+ AbstractLifeCycle.LOGGER.error("No layout provided for KafkaAppender");
+ return null;
+ }
+ final KafkaManager kafkaManager = KafkaManager.getManager(configuration.getLoggerContext(), name, topic, true,
+ properties, key);
+ return new KafkaAppender(name, layout, filter, ignoreExceptions, kafkaManager, null, null);
+ }
+
+ /**
+ * Tests if the given log event is from a Kafka Producer implementation.
+ *
+ * @param event The event to test.
+ * @return true to avoid recursion and skip logging, false to log.
+ */
+ private static boolean isRecursive(final LogEvent event) {
+ return Stream.of(KAFKA_CLIENT_PACKAGES).anyMatch(prefix -> event.getLoggerName().startsWith(prefix));
+ }
+
+ /**
+ * Creates a builder for a KafkaAppender.
+ *
+ * @return a builder for a KafkaAppender.
+ */
+ @PluginBuilderFactory
+ public static <B extends Builder<B>> B newBuilder() {
+ return new Builder<B>().asBuilder();
+ }
+
+ private final Integer retryCount;
+
+ private final KafkaManager manager;
+
+ private KafkaAppender(final String name, final Layout<? extends Serializable> layout, final Filter filter,
+ final boolean ignoreExceptions, final KafkaManager manager, final Property[] properties,
+ final Integer retryCount) {
+ super(name, filter, layout, ignoreExceptions, properties);
+ this.manager = Objects.requireNonNull(manager, "manager");
+ this.retryCount = retryCount;
+ }
+
+ @Override
+ public void append(final LogEvent event) {
+ if (event.getLoggerName() != null && isRecursive(event)) {
+ LOGGER.warn("Recursive logging from [{}] for appender [{}].", event.getLoggerName(), getName());
+ } else {
+ try {
+ tryAppend(event);
+ } catch (final Exception e) {
+
+ if (this.retryCount != null) {
+ int currentRetryAttempt = 0;
+ while (currentRetryAttempt < this.retryCount) {
+ currentRetryAttempt++;
+ try {
+ tryAppend(event);
+ break;
+ } catch (Exception e1) {
+
+ }
+ }
+ }
+ error("Unable to write to Kafka in appender [" + getName() + "]", event, e);
+ }
+ }
+ }
+
+ @Override
+ public void start() {
+ super.start();
+ manager.startup();
}
- /**
- * Creates a builder for a KafkaAppender.
- *
- * @return a builder for a KafkaAppender.
- */
- @PluginBuilderFactory
- public static <B extends Builder<B>> B newBuilder() {
- return new Builder<B>().asBuilder();
- }
-
- private final Integer retryCount;
-
- private final KafkaManager manager;
-
- private KafkaAppender(final String name, final Layout<? extends Serializable> layout, final Filter filter,
- final boolean ignoreExceptions, final KafkaManager manager, final Property[] properties,
- final Integer retryCount) {
- super(name, filter, layout, ignoreExceptions, properties);
- this.manager = Objects.requireNonNull(manager, "manager");
- this.retryCount = retryCount;
- }
-
- @Override
- public void append(final LogEvent event) {
- if (event.getLoggerName() != null && isRecursive(event)) {
- LOGGER.warn("Recursive logging from [{}] for appender [{}].", event.getLoggerName(), getName());
- } else {
- try {
- tryAppend(event);
- } catch (final Exception e) {
-
- if (this.retryCount != null) {
- int currentRetryAttempt = 0;
- while (currentRetryAttempt < this.retryCount) {
- currentRetryAttempt++;
- try {
- tryAppend(event);
- break;
- } catch (Exception e1) {
-
- }
- }
- }
- error("Unable to write to Kafka in appender [" + getName() + "]", event, e);
- }
- }
- }
-
- @Override
- public void start() {
- super.start();
- manager.startup();
- }
-
- @Override
- public boolean stop(final long timeout, final TimeUnit timeUnit) {
- setStopping();
- boolean stopped = super.stop(timeout, timeUnit, false);
- stopped &= manager.stop(timeout, timeUnit);
- setStopped();
- return stopped;
- }
-
- @Override
- public String toString() {
- return "KafkaAppender{" + "name=" + getName() + ", state=" + getState() + ", topic=" + manager.getTopic() + '}';
- }
-
- private void tryAppend(final LogEvent event) throws ExecutionException, InterruptedException, TimeoutException {
- final Layout<? extends Serializable> layout = getLayout();
- byte[] data;
- if (layout instanceof SerializedLayout) {
- final byte[] header = layout.getHeader();
- final byte[] body = layout.toByteArray(event);
- data = new byte[header.length + body.length];
- System.arraycopy(header, 0, data, 0, header.length);
- System.arraycopy(body, 0, data, header.length, body.length);
- } else {
- data = layout.toByteArray(event);
- }
- manager.send(data);
- }
+ @Override
+ public boolean stop(final long timeout, final TimeUnit timeUnit) {
+ setStopping();
+ boolean stopped = super.stop(timeout, timeUnit, false);
+ stopped &= manager.stop(timeout, timeUnit);
+ setStopped();
+ return stopped;
+ }
+
+ @Override
+ public String toString() {
+ return "KafkaAppender{" + "name=" + getName() + ", state=" + getState() + ", topic=" + manager.getTopic() + '}';
+ }
+
+ private void tryAppend(final LogEvent event) throws ExecutionException, InterruptedException, TimeoutException {
+ final Layout<? extends Serializable> layout = getLayout();
+ byte[] data;
+ if (layout instanceof SerializedLayout) {
+ final byte[] header = layout.getHeader();
+ final byte[] body = layout.toByteArray(event);
+ data = new byte[header.length + body.length];
+ System.arraycopy(header, 0, data, 0, header.length);
+ System.arraycopy(body, 0, data, header.length, body.length);
+ } else {
+ data = layout.toByteArray(event);
+ }
+ manager.send(data);
+ }
}
diff --git a/log4j-core/src/main/java/org/apache/logging/log4j/core/appender/mom/kafka/KafkaManager.java b/log4j-core/src/main/java/org/apache/logging/log4j/core/appender/mom/kafka/KafkaManager.java
index 14a1dff9f3..378099e734 100644
--- a/log4j-core/src/main/java/org/apache/logging/log4j/core/appender/mom/kafka/KafkaManager.java
+++ b/log4j-core/src/main/java/org/apache/logging/log4j/core/appender/mom/kafka/KafkaManager.java
@@ -14,7 +14,6 @@
* See the license for the specific language governing permissions and
* limitations under the license.
*/
-
package org.apache.logging.log4j.core.appender.mom.kafka;
import java.nio.charset.StandardCharsets;
@@ -37,143 +36,143 @@ import org.apache.logging.log4j.core.util.Log4jThread;
public class KafkaManager extends AbstractManager {
- public static final String DEFAULT_TIMEOUT_MILLIS = "30000";
-
- /**
- * package-private access for testing.
- */
- static KafkaProducerFactory producerFactory = new DefaultKafkaProducerFactory();
-
- private final Properties config = new Properties();
- private Producer<byte[], byte[]> producer;
- private final int timeoutMillis;
-
- private final String topic;
- private final String key;
- private final boolean syncSend;
- private static final KafkaManagerFactory factory = new KafkaManagerFactory();
-
- /*
- * The Constructor should have been declared private as all Managers are create
- * by the internal factory;
- */
- public KafkaManager(final LoggerContext loggerContext, final String name, final String topic,
- final boolean syncSend, final Property[] properties, final String key) {
- super(loggerContext, name);
- this.topic = Objects.requireNonNull(topic, "topic");
- this.syncSend = syncSend;
-
- config.setProperty("key.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
- config.setProperty("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
- config.setProperty("batch.size", "0");
-
- for (final Property property : properties) {
- config.setProperty(property.getName(), property.getValue());
- }
-
- this.key = key;
-
- this.timeoutMillis = Integers.parseInt(config.getProperty("timeout.ms", DEFAULT_TIMEOUT_MILLIS));
- }
-
- @Override
- public boolean releaseSub(final long timeout, final TimeUnit timeUnit) {
- if (timeout > 0) {
- closeProducer(timeout, timeUnit);
- } else {
- closeProducer(timeoutMillis, TimeUnit.MILLISECONDS);
- }
- return true;
- }
-
- private void closeProducer(final long timeout, final TimeUnit timeUnit) {
- if (producer != null) {
- // This thread is a workaround for this Kafka issue:
- // https://issues.apache.org/jira/browse/KAFKA-1660
- final Thread closeThread = new Log4jThread(() -> {
- if (producer != null) {
- producer.close();
- }
+ public static final String DEFAULT_TIMEOUT_MILLIS = "30000";
+
+ /**
+ * package-private access for testing.
+ */
+ static KafkaProducerFactory producerFactory = new DefaultKafkaProducerFactory();
+
+ private final Properties config = new Properties();
+ private Producer<byte[], byte[]> producer;
+ private final int timeoutMillis;
+
+ private final String topic;
+ private final String key;
+ private final boolean syncSend;
+ private static final KafkaManagerFactory factory = new KafkaManagerFactory();
+
+ /*
+ * The Constructor should have been declared private as all Managers are create
+ * by the internal factory;
+ */
+ public KafkaManager(final LoggerContext loggerContext, final String name, final String topic,
+ final boolean syncSend, final Property[] properties, final String key) {
+ super(loggerContext, name);
+ this.topic = Objects.requireNonNull(topic, "topic");
+ this.syncSend = syncSend;
+
+ config.setProperty("key.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
+ config.setProperty("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
+ config.setProperty("batch.size", "0");
+
+ for (final Property property : properties) {
+ config.setProperty(property.getName(), property.getValue());
+ }
+
+ this.key = key;
+
+ this.timeoutMillis = Integers.parseInt(config.getProperty("timeout.ms", DEFAULT_TIMEOUT_MILLIS));
+ }
+
+ @Override
+ public boolean releaseSub(final long timeout, final TimeUnit timeUnit) {
+ if (timeout > 0) {
+ closeProducer(timeout, timeUnit);
+ } else {
+ closeProducer(timeoutMillis, TimeUnit.MILLISECONDS);
+ }
+ return true;
+ }
+
+ private void closeProducer(final long timeout, final TimeUnit timeUnit) {
+ if (producer != null) {
+ // This thread is a workaround for this Kafka issue:
+ // https://issues.apache.org/jira/browse/KAFKA-1660
+ final Thread closeThread = new Log4jThread(() -> {
+ if (producer != null) {
+ producer.close();
+ }
}, "KafkaManager-CloseThread");
- closeThread.setDaemon(true); // avoid blocking JVM shutdown
- closeThread.start();
- try {
- closeThread.join(timeUnit.toMillis(timeout));
- } catch (final InterruptedException ignore) {
- Thread.currentThread().interrupt();
- // ignore
- }
- }
- }
-
- public void send(final byte[] msg) throws ExecutionException, InterruptedException, TimeoutException {
- if (producer != null) {
- byte[] newKey = null;
-
- if (key != null && key.contains("${")) {
- newKey = getLoggerContext().getConfiguration().getStrSubstitutor().replace(key)
- .getBytes(StandardCharsets.UTF_8);
- } else if (key != null) {
- newKey = key.getBytes(StandardCharsets.UTF_8);
- }
-
- final ProducerRecord<byte[], byte[]> newRecord = new ProducerRecord<>(topic, newKey, msg);
- if (syncSend) {
- final Future<RecordMetadata> response = producer.send(newRecord);
- response.get(timeoutMillis, TimeUnit.MILLISECONDS);
- } else {
- producer.send(newRecord, (metadata, e) -> {
- if (e != null) {
- LOGGER.error("Unable to write to Kafka in appender [" + getName() + "]", e);
- }
+ closeThread.setDaemon(true); // avoid blocking JVM shutdown
+ closeThread.start();
+ try {
+ closeThread.join(timeUnit.toMillis(timeout));
+ } catch (final InterruptedException ignore) {
+ Thread.currentThread().interrupt();
+ // ignore
+ }
+ }
+ }
+
+ public void send(final byte[] msg) throws ExecutionException, InterruptedException, TimeoutException {
+ if (producer != null) {
+ byte[] newKey = null;
+
+ if (key != null && key.contains("${")) {
+ newKey = getLoggerContext().getConfiguration().getStrSubstitutor().replace(key)
+ .getBytes(StandardCharsets.UTF_8);
+ } else if (key != null) {
+ newKey = key.getBytes(StandardCharsets.UTF_8);
+ }
+
+ final ProducerRecord<byte[], byte[]> newRecord = new ProducerRecord<>(topic, newKey, msg);
+ if (syncSend) {
+ final Future<RecordMetadata> response = producer.send(newRecord);
+ response.get(timeoutMillis, TimeUnit.MILLISECONDS);
+ } else {
+ producer.send(newRecord, (metadata, e) -> {
+ if (e != null) {
+ LOGGER.error("Unable to write to Kafka in appender [" + getName() + "]", e);
+ }
});
- }
- }
- }
-
- public void startup() {
- if (producer == null) {
- producer = producerFactory.newKafkaProducer(config);
- }
- }
-
- public String getTopic() {
- return topic;
- }
-
- public static KafkaManager getManager(final LoggerContext loggerContext, final String name, final String topic,
- final boolean syncSend, final Property[] properties, final String key) {
- StringBuilder sb = new StringBuilder(name);
- sb.append(" ").append(topic).append(" ").append(syncSend + "");
- for (Property prop : properties) {
- sb.append(" ").append(prop.getName()).append("=").append(prop.getValue());
- }
- return getManager(sb.toString(), factory, new FactoryData(loggerContext, topic, syncSend, properties, key));
- }
-
- private static class FactoryData {
- private final LoggerContext loggerContext;
- private final String topic;
- private final boolean syncSend;
- private final Property[] properties;
- private final String key;
-
- public FactoryData(final LoggerContext loggerContext, final String topic, final boolean syncSend,
- final Property[] properties, final String key) {
- this.loggerContext = loggerContext;
- this.topic = topic;
- this.syncSend = syncSend;
- this.properties = properties;
- this.key = key;
- }
-
- }
-
- private static class KafkaManagerFactory implements ManagerFactory<KafkaManager, FactoryData> {
- @Override
- public KafkaManager createManager(String name, FactoryData data) {
- return new KafkaManager(data.loggerContext, name, data.topic, data.syncSend, data.properties, data.key);
- }
- }
+ }
+ }
+ }
+
+ public void startup() {
+ if (producer == null) {
+ producer = producerFactory.newKafkaProducer(config);
+ }
+ }
+
+ public String getTopic() {
+ return topic;
+ }
+
+ public static KafkaManager getManager(final LoggerContext loggerContext, final String name, final String topic,
+ final boolean syncSend, final Property[] properties, final String key) {
+ StringBuilder sb = new StringBuilder(name);
+ sb.append(" ").append(topic).append(" ").append(syncSend + "");
+ for (Property prop : properties) {
+ sb.append(" ").append(prop.getName()).append("=").append(prop.getValue());
+ }
+ return getManager(sb.toString(), factory, new FactoryData(loggerContext, topic, syncSend, properties, key));
+ }
+
+ private static class FactoryData {
+ private final LoggerContext loggerContext;
+ private final String topic;
+ private final boolean syncSend;
+ private final Property[] properties;
+ private final String key;
+
+ public FactoryData(final LoggerContext loggerContext, final String topic, final boolean syncSend,
+ final Property[] properties, final String key) {
+ this.loggerContext = loggerContext;
+ this.topic = topic;
+ this.syncSend = syncSend;
+ this.properties = properties;
+ this.key = key;
+ }
+
+ }
+
+ private static class KafkaManagerFactory implements ManagerFactory<KafkaManager, FactoryData> {
+ @Override
+ public KafkaManager createManager(String name, FactoryData data) {
+ return new KafkaManager(data.loggerContext, name, data.topic, data.syncSend, data.properties, data.key);
+ }
+ }
}