feat: add KSQL processing log message on uncaught streams exceptions #6253

spena · 2020-09-18T03:41:44Z

Description

Calls the ProcessingLogger on uncaught errors by Kafka streams. The processing log uncaught handler is registered by the PersistentQueryMetadata only. Transient queries do not need to log errors to the processing log (they're at sent back to the user instead).

The error sent to the logger uses a new KafkaStreamsThreadErrorMessage type that creates an structured error message. It contains the error message, the thread and the exception.

Testing done

Added unit tests

ksql> select * from KSQL_PROCESSING_LOG emit changes;
+---------------------------------------------+---------------------------------------------+---------------------------------------------+---------------------------------------------+
|LOGGER                                       |LEVEL                                        |TIME                                         |MESSAGE                                      |
+---------------------------------------------+---------------------------------------------+---------------------------------------------+---------------------------------------------+
|processing.CSAS_S1_0.ksql.logger.thread.excep|ERROR                                        |1600872731171                                |{TYPE=4, DESERIALIZATIONERROR=null, RECORDPRO|
|tion.uncaught                                |                                             |                                             |CESSINGERROR=null, PRODUCTIONERROR=null, SERI|
|                                             |                                             |                                             |ALIZATIONERROR=null, KAFKASTREAMSTHREADERROR=|
|                                             |                                             |                                             |{ERRORMESSAGE=Unhandled exception caught in s|
|                                             |                                             |                                             |treams thread, THREADNAME=_confluent-ksql-def|
|                                             |                                             |                                             |ault_query_CSAS_S1_0-22278e08-e1ab-475f-84fe-|
|                                             |                                             |                                             |c4c97815c5d4-StreamThread-4, CAUSE=[One or mo|
|                                             |                                             |                                             |re source topics were missing during rebalanc|
|                                             |                                             |                                             |e]}}                                         |

Reviewer checklist

Ensure docs are updated if necessary. (eg. if a user visible feature is being added or changed).
Ensure relevant issues are linked (description should include text like "Fixes #")

rodesai

Looking good. Why are we only doing this for persistent queries?

rodesai · 2020-09-22T06:12:19Z

...db-common/src/main/java/io/confluent/ksql/logging/processing/ProcessingLogMessageSchema.java

@@ -76,11 +76,23 @@
      .optional()
      .build();

+  public static final String KAFKA_STREAMS_THREAD_ERROR_FIELD_MESSAGE = "errorMessage";
+  public static final String KAFKA_STREAMS_THREAD_ERROR_FIELD_NAME = "name";


let's use the name "threadName"

rodesai · 2020-09-22T06:22:47Z

ksqldb-engine/src/main/java/io/confluent/ksql/query/QueryExecutor.java

@@ -229,7 +229,9 @@ public PersistentQueryMetadata buildPersistentQuery(
        ksqlConfig.getLong(KSQL_SHUTDOWN_TIMEOUT_MS_CONFIG),
        classifier,
        physicalPlan,
-        ksqlConfig.getInt(KsqlConfig.KSQL_QUERY_ERROR_MAX_QUEUE_SIZE)
+        ksqlConfig.getInt(KsqlConfig.KSQL_QUERY_ERROR_MAX_QUEUE_SIZE),
+        (ProcessingLogger) streamsProperties.get(


We should use a different logger here. The logger is important because its name tells you a lot about the context of the error. In this case, the logger's name is ksql.logger.production.error, which tells you this is an error hit while trying to produce a record to Kafka. Other loggers' names include the node in the query topology. Eventually we'll let users adjust the log levels for some loggers at runtime, so you could see very detailed logging from just 1 UDF for example. In this case I'd name this something like ksql.logger.thread.exception.uncaught.

To create the logger, you can use the factory in the processingLogContext member of this class.

rodesai · 2020-09-22T06:24:07Z

ksqldb-engine/src/main/java/io/confluent/ksql/util/PersistentQueryMetadata.java

+  protected void uncaughtHandler(final Thread thread, final Throwable error) {
+    super.uncaughtHandler(thread, error);
+
+    processingLogger.error(KafkaStreamsThreadError.of(


consider logging this at the fatal level, since this is a thread-killing error.

How do I log at fatal level?

The ProcessingLogger has only error, an the inner logger has only error, info, debug.

spena · 2020-09-22T15:40:30Z

@rodesai I didn't do transient queries because seems they catch the uncaught errors differently (not by QueryMetadata::uncaughtHandler), and they send the error back to the client.

I assumed this is what we want, that users who are running push/pull queries see the error in their side 'cause they can do it. Contrary to persistent queries that they cannot see it unless the look in the processing log. Does it make sense? or should we also configure transient queries to use the internal QueryMetadata::uncaughtHandler() and processing log handler too?

rodesai · 2020-09-22T20:49:40Z

@rodesai I didn't do transient queries because seems they catch the uncaught errors differently (not by QueryMetadata::uncaughtHandler), and they send the error back to the client.

This makes sense. Thanks for the explainer.

JimGalasyn

LGTM!

spena · 2020-09-23T14:59:08Z

@rodesai I added the queryId to the logger name. This is how it looks now:

ksql> select * from KSQL_PROCESSING_LOG emit changes;
+---------------------------------------------+---------------------------------------------+---------------------------------------------+---------------------------------------------+
|LOGGER                                       |LEVEL                                        |TIME                                         |MESSAGE                                      |
+---------------------------------------------+---------------------------------------------+---------------------------------------------+---------------------------------------------+
|processing.CSAS_S1_0.ksql.logger.thread.excep|ERROR                                        |1600872731171                                |{TYPE=4, DESERIALIZATIONERROR=null, RECORDPRO|
|tion.uncaught                                |                                             |                                             |CESSINGERROR=null, PRODUCTIONERROR=null, SERI|
|                                             |                                             |                                             |ALIZATIONERROR=null, KAFKASTREAMSTHREADERROR=|
|                                             |                                             |                                             |{ERRORMESSAGE=Unhandled exception caught in s|
|                                             |                                             |                                             |treams thread, THREADNAME=_confluent-ksql-def|
|                                             |                                             |                                             |ault_query_CSAS_S1_0-22278e08-e1ab-475f-84fe-|
|                                             |                                             |                                             |c4c97815c5d4-StreamThread-4, CAUSE=[One or mo|
|                                             |                                             |                                             |re source topics were missing during rebalanc|
|                                             |                                             |                                             |e]}}                                         |

agavra

LGTM pending Rohan's comments

rodesai

LGTM

spena added enhancement P0 Denotes must-have for a given milestone labels Sep 18, 2020

spena added this to the 0.13.0 milestone Sep 18, 2020

spena requested review from rodesai and a team September 18, 2020 03:41

spena force-pushed the add_processing_log_msg branch 2 times, most recently from 9219c6d to 430c7ac Compare September 21, 2020 14:36

rodesai reviewed Sep 22, 2020

View reviewed changes

spena requested a review from JimGalasyn as a code owner September 22, 2020 16:50

spena added 2 commits September 22, 2020 15:34

feat: add KSQL processing log message on uncaught streams exceptions

91d9b35

fix: address PR feedback

79b471c

spena force-pushed the add_processing_log_msg branch from d03ddd5 to 79b471c Compare September 22, 2020 20:35

JimGalasyn approved these changes Sep 22, 2020

View reviewed changes

fix: add queryId to logger name

56a28eb

agavra approved these changes Sep 23, 2020

View reviewed changes

rodesai approved these changes Sep 23, 2020

View reviewed changes

spena merged commit ac8875f into confluentinc:master Sep 23, 2020

spena deleted the add_processing_log_msg branch September 23, 2020 21:12

jzaralim mentioned this pull request Oct 1, 2020

Update some of the outputs for KSQL v0.13.0 confluentinc/kafka-tutorials#569

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: add KSQL processing log message on uncaught streams exceptions #6253

feat: add KSQL processing log message on uncaught streams exceptions #6253

spena commented Sep 18, 2020 •

edited by agavra

Loading

rodesai left a comment

rodesai Sep 22, 2020

spena Sep 22, 2020

rodesai Sep 22, 2020

spena Sep 22, 2020

rodesai Sep 22, 2020

spena Sep 22, 2020

spena commented Sep 22, 2020

rodesai commented Sep 22, 2020

JimGalasyn left a comment

spena commented Sep 23, 2020

agavra left a comment

rodesai left a comment

feat: add KSQL processing log message on uncaught streams exceptions #6253

feat: add KSQL processing log message on uncaught streams exceptions #6253

Conversation

spena commented Sep 18, 2020 • edited by agavra Loading

Description

Testing done

Reviewer checklist

rodesai left a comment

Choose a reason for hiding this comment

rodesai Sep 22, 2020

Choose a reason for hiding this comment

spena Sep 22, 2020

Choose a reason for hiding this comment

rodesai Sep 22, 2020

Choose a reason for hiding this comment

spena Sep 22, 2020

Choose a reason for hiding this comment

rodesai Sep 22, 2020

Choose a reason for hiding this comment

spena Sep 22, 2020

Choose a reason for hiding this comment

spena commented Sep 22, 2020

rodesai commented Sep 22, 2020

JimGalasyn left a comment

Choose a reason for hiding this comment

spena commented Sep 23, 2020

agavra left a comment

Choose a reason for hiding this comment

rodesai left a comment

Choose a reason for hiding this comment

spena commented Sep 18, 2020 •

edited by agavra

Loading