[
https://issues.apache.org/jira/browse/KAFKA-12774?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17352499#comment-17352499
]
Almog Gavra commented on KAFKA-12774:
-------------------------------------
I went back to triage the severity of this because if things aren't getting
logged via Log4j then our applications would be missing log-shipping, which is
crticial in our production environments.
To confirm nothing was being logged outside of Log4J I hacked together
something to try to reproduce this and I also added the following line any time
a record was to be sent in RecordCollectorImpl to make sure I had exactly the
same type of error that you encountered:
{code:java}
recordSendError(topic, new InvalidPidMappingException("foo"), serializedRecord);
{code}
I could not reproduce. The only things that were logged (notice that all Log4j
loggers are turned OFF so only things that get logged to stdout get logged)
were what I specifially logged to stdout. This was the stdout (I printed to
stdout in the uncaught handler instead of logging):
{code:java}
SEEN: 0,0
(HERE) Uncaught exception handled - replacing thread Error encountered sending
record to topic bar for task 0_0 due to:
org.apache.kafka.common.errors.InvalidPidMappingException: foo
Exception handler choose to FAIL the processing, no more records would be sent.
{code}
We can leave this ticket open in case anyone else has any Ideas.
----
Here is the app:
{code:java}
public static void main(String[] args) throws InterruptedException, IOException
{
LogManager.getRootLogger().setLevel(Level.OFF);
@SuppressWarnings("unchecked") Enumeration<Logger> loggers =
LogManager.getCurrentLoggers();
while (loggers.hasMoreElements()) {
loggers.nextElement().setLevel(Level.OFF);
}
final EmbeddedKafkaCluster cluster = new EmbeddedKafkaCluster(1);
cluster.start();
cluster.createTopic("foo");
final Properties config = new Properties();
config.put(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG,
StreamsConfig.OPTIMIZE);
config.put(StreamsConfig.APPLICATION_ID_CONFIG, "app");
config.put(StreamsConfig.APPLICATION_SERVER_CONFIG, "localhost:" + 123);
config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
cluster.bootstrapServers());
config.put(StreamsConfig.STATE_DIR_CONFIG,
TestUtils.tempDirectory().getPath());
config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG,
Serdes.Integer().getClass());
config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,
Serdes.Integer().getClass());
config.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 1);
config.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100);
config.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 200);
config.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 1000);
config.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 100);
final CountDownLatch sawKey = new CountDownLatch(1);
final StreamsBuilder builder = new StreamsBuilder();
builder.stream("foo")
.filter((k, v) -> k != null)
.peek((k, v) -> System.out.println("SEEN: " + k + "," + v))
.peek((k ,v) -> {
if ((int) k == 0) sawKey.countDown();
})
.to("bar");
final Topology build = builder.build(config);
final KafkaStreams app = new KafkaStreams(build, config);
app.setUncaughtExceptionHandler(exception -> {
System.out.println("(HERE) Uncaught exception handled - replacing
thread " + exception.getMessage());
return
StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.REPLACE_THREAD;
}
);
final CountDownLatch startLatch = new CountDownLatch(1);
app.setStateListener((newState, oldState) -> {
if (newState == State.RUNNING) {
startLatch.countDown();
}
});
app.start();
startLatch.await();
final Properties producerProps = new Properties();
producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
cluster.bootstrapServers());
producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
IntegerSerializer.class);
producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
IntegerSerializer.class);
IntegrationTestUtils.produceKeyValuesSynchronously(
"foo",
IntStream.range(0, 1)
.mapToObj(i -> KeyValue.pair(0, i))
.collect(Collectors.toList()),
producerProps,
Time.SYSTEM);
sawKey.await();
app.close();
app.cleanUp();
cluster.after();
}{code}
> kafka-streams 2.8: logging in uncaught-exceptionhandler doesn't go through
> log4j
> --------------------------------------------------------------------------------
>
> Key: KAFKA-12774
> URL: https://issues.apache.org/jira/browse/KAFKA-12774
> Project: Kafka
> Issue Type: Bug
> Components: streams
> Affects Versions: 2.8.0
> Reporter: Jørgen
> Priority: Minor
> Fix For: 3.0.0, 2.8.1
>
>
> When exceptions is handled in the uncaught-exception handler introduced in
> KS2.8, the logging of the stacktrace doesn't seem to go through the logging
> framework configured by the application (log4j2 in our case), but gets
> printed to console "line-by-line".
> All other exceptions logged by kafka-streams go through log4j2 and gets
> formatted properly according to the log4j2 appender (json in our case).
> Haven't tested this on other frameworks like logback.
> Application setup:
> * Spring-boot 2.4.5
> * Log4j 2.13.3
> * Slf4j 1.7.30
> Log4j2 appender config:
> {code:java}
> <Appenders>
> <Console name="Console" target="SYSTEM_OUT">
> <JSONLayout complete="false" compact="true" eventEol="true"
> stacktraceAsString="true" properties="true">
> <KeyValuePair key="timestamp"
> value="$${date:yyyy-MM-dd'T'HH:mm:ss.SSSZ}"/>
> </JSONLayout>
> </Console>
> </Appenders> {code}
> Uncaught exception handler config:
> {code:java}
> kafkaStreams.setUncaughtExceptionHandler { exception ->
> logger.warn("Uncaught exception handled - replacing thread", exception)
> // logged properly
>
> StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.REPLACE_THREAD
> } {code}
> Stacktrace that gets printed line-by-line:
> {code:java}
> Exception in thread "xxx-f5860dff-9a41-490e-8ab0-540b1a7f9ce4-StreamThread-2"
> org.apache.kafka.streams.errors.StreamsException: Error encountered sending
> record to topic xxx-repartition for task 3_2 due
> to:org.apache.kafka.common.errors.InvalidPidMappingException: The producer
> attempted to use a producer id which is not currently assigned to its
> transactional id.Exception handler choose to FAIL the processing, no more
> records would be sent. at
> org.apache.kafka.streams.processor.internals.RecordCollectorImpl.recordSendError(RecordCollectorImpl.java:226)
> at
> org.apache.kafka.streams.processor.internals.RecordCollectorImpl.lambda$send$0(RecordCollectorImpl.java:196)
> at
> org.apache.kafka.clients.producer.KafkaProducer$InterceptorCallback.onCompletion(KafkaProducer.java:1365)
> at
> org.apache.kafka.clients.producer.internals.ProducerBatch.completeFutureAndFireCallbacks(ProducerBatch.java:231)
> at
> org.apache.kafka.clients.producer.internals.ProducerBatch.abort(ProducerBatch.java:159)
> at
> org.apache.kafka.clients.producer.internals.RecordAccumulator.abortUndrainedBatches(RecordAccumulator.java:783)
> at
> org.apache.kafka.clients.producer.internals.Sender.maybeSendAndPollTransactionalRequest(Sender.java:430)
> at
> org.apache.kafka.clients.producer.internals.Sender.runOnce(Sender.java:315)
> at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:242)
> at java.base/java.lang.Thread.run(Unknown Source)Caused by:
> org.apache.kafka.common.errors.InvalidPidMappingException: The producer
> attempted to use a producer id which is not currently assigned to its
> transactional id. {code}
>
> It's a little bit hard to reproduce as I haven't found any way to trigger
> uncaught-exception-handler through junit-tests.
> Link to discussion on slack:
> https://confluentcommunity.slack.com/archives/C48AHTCUQ/p1620389197436700
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
