Introduce Kafka event logger for server events #4733

bowenliang123 · 2023-04-19T02:46:30Z

Why are the changes needed?

introduce new event logger type KAFKA
send server events to the Kafka topic with initializing and closing Kafka producer properly with server's lifecyle
use Kafka 3.4.0 as the client version, and tested with Kakfa servers of 2.8.x and 3.4.x

How was this patch tested?

Add some test cases that check the changes thoroughly including negative and positive cases if possible
Add screenshots for manual tests if appropriate
Run test locally before make a pull request

bowenliang123 · 2023-04-19T03:02:10Z

cc @lxbalex @jeanlyn

codecov-commenter · 2023-04-19T05:05:27Z

Codecov Report

Merging #4733 (b5220d2) into master (6ae0c8b) will increase coverage by 0.06%.
The diff coverage is 93.22%.

@@             Coverage Diff              @@
##             master    #4733      +/-   ##
============================================
+ Coverage     57.98%   58.05%   +0.06%     
  Complexity       13       13              
============================================
  Files           581      583       +2     
  Lines         32491    32548      +57     
  Branches       4315     4317       +2     
============================================
+ Hits          18841    18896      +55     
+ Misses        11836    11834       -2     
- Partials       1814     1818       +4

Impacted Files	Coverage Δ
...rg/apache/kyuubi/events/EventHandlerRegister.scala	`50.00% <33.33%> (-2.39%)`	⬇️
...vents/handler/ServerKafkaLoggingEventHandler.scala	`88.88% <88.88%> (ø)`
...che/kyuubi/events/ServerEventHandlerRegister.scala	`94.73% <90.00%> (-5.27%)`	⬇️
...in/scala/org/apache/kyuubi/config/KyuubiConf.scala	`97.20% <100.00%> (+0.02%)`	⬆️
...ala/org/apache/kyuubi/events/EventLoggerType.scala	`100.00% <100.00%> (ø)`
...uubi/events/handler/KafkaLoggingEventHandler.scala	`100.00% <100.00%> (ø)`

... and 5 files with indirect coverage changes

📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more

pan3793 · 2023-04-20T10:07:45Z

We can use Testcontianers to address the Kafka server compatibility issues

pan3793 · 2023-04-20T10:08:28Z

kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala

  val SERVER_EVENT_LOGGERS: ConfigEntry[Seq[String]] =
    buildConf("kyuubi.backend.server.event.loggers")
      .doc("A comma-separated list of server history loggers, where session/operation etc" +
        " events go.<ul>" +
        s" <li>JSON: the events will be written to the location of" +
        s" ${SERVER_EVENT_JSON_LOG_PATH.key}</li>" +
+        s" <li>" +
+        s"KAFKA: the events sent to topic of `${SERVER_EVENT_KAFKA_TOPIC.key}`, " +


unify the format

it's unclear what's content in the kafka record

OK, change to mention the format with KAFKA: the events serialized in JSON format

pan3793 · 2023-04-20T10:09:01Z

dev/dependencyList

@@ -181,6 +183,7 @@ simpleclient_tracer_otel/0.16.0//simpleclient_tracer_otel-0.16.0.jar
 simpleclient_tracer_otel_agent/0.16.0//simpleclient_tracer_otel_agent-0.16.0.jar
 slf4j-api/1.7.36//slf4j-api-1.7.36.jar
 snakeyaml/1.33//snakeyaml-1.33.jar
+snappy-java/1.1.7.3//snappy-java-1.1.7.3.jar


does it support arm?

when introduce new artifacts in release tarball, don't forget to update LICENSE/NOTICE

snappy-java supports Windows, Mac OS X, Linux (x86, x86_64, arm, etc...). If your platform is not supported, you need to build native libraries by yourself.

Accorrding to snappy-java's doc here(https://github.com/xerial/snappy-java/blob/master/BUILD.md), arm architecture should be supported.

kyuubi-server/src/main/scala/org/apache/kyuubi/events/ServerEventHandlerRegister.scala

...er/src/test/scala/org/apache/kyuubi/events/handler/ServerKafkaLoggingEventHandlerSuite.scala

pan3793 · 2023-04-20T10:14:49Z

kyuubi-events/src/main/scala/org/apache/kyuubi/events/handler/KafkaLoggingEventHandler.scala

+
+  override def apply(event: KyuubiEvent): Unit = {
+    try {
+      val record = new ProducerRecord[String, String](topic, event.eventType, event.toJson)


I remember that Kafka allows setting event timestamp on constructing ProducerRecord, it may be useful in some cases.

Good suggestion. Setting timestamp with System.currentTimeMillis. Is that good enough? Or use Java8's Instant.now.toEpochMilli (with UTC clock)?

While timestamp is not set , the producer will assign the timestamp using System.currentTimeMillis().

Usually, it should be event time, but I just notice that KyuubiEvent does not have event time ...

Yes, no available unified event time in KyuubiEvent right now.

let's leave a TODO here, we should add an event_time in KyuubiEvent and propagate it to Kafka, but it may break the backward compatibility, so better to do it later separately

kyuubi-events/src/main/scala/org/apache/kyuubi/events/handler/KafkaLoggingEventHandler.scala

pom.xml

bowenliang123 · 2023-04-25T14:03:50Z

cc @iodone @wForget

I am looking for proper timing to close event handlers for Kyuubi Server. But repeated errors comes in the irreverent places.

As I tried to close event handlers via EventBus.close() in KyuubiSever's stop() or stopServer,
the get operation log test case failed, which is completely unrelated to Kafka logger introduced in this PR, and it also never uses the Kafka logger in context.

If you guys have any idea for this, feel free to ping me any time. Thanks.

ERROR org.apache.kyuubi.events.handler.ServerKafkaLoggingEventHandler: Failed to send event in KafkaEventHandler
java.lang.IllegalStateException: Cannot perform operation after producer has been closed
	at org.apache.kafka.clients.producer.KafkaProducer.throwIfProducerClosed(KafkaProducer.java:892) ~[kafka-clients-2.8.1.jar:?]
	at org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:901) ~[kafka-clients-2.8.1.jar:?]
	at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:885) ~[kafka-clients-2.8.1.jar:?]
	at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:773) ~[kafka-clients-2.8.1.jar:?]
	at org.apache.kyuubi.events.handler.KafkaLoggingEventHandler.apply(KafkaLoggingEventHandler.scala:50) ~[kyuubi-events_2.12-1.8.0-SNAPSHOT.jar:?]
	at org.apache.kyuubi.events.handler.KafkaLoggingEventHandler.apply(KafkaLoggingEventHandler.scala:31) ~[kyuubi-events_2.12-1.8.0-SNAPSHOT.jar:?]
	at org.apache.kyuubi.events.EventBus$EventBusLive.$anonfun$post$4(EventBus.scala:87) ~[kyuubi-events_2.12-1.8.0-SNAPSHOT.jar:?]
	at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23) ~[scala-library-2.12.17.jar:?]
- get operation log *** FAILED ***
  scala.collection.JavaConverters.asScalaBufferConverter[String](logRowSet.getLogRowSet()).asScala.exists(((x$1: String) => x$1.contains("show tables"))) was false (OperationsResourceSuite.scala:107)

see : https://github.com/apache/kyuubi/actions/runs/4795357563/jobs/8529805995?pr=4733#step:8:1778

...ver/src/test/scala/org/apache/kyuubi/events/handler/ServerJsonLoggingEventHandlerSuite.scala

cxzl25 · 2023-05-04T11:30:29Z

kyuubi-events/src/main/scala/org/apache/kyuubi/events/handler/KafkaLoggingEventHandler.scala

+
+  override def close(): Unit = {
+    kafkaProducer.flush()
+    kafkaProducer.close()


I suggest to use close method with timeout.

org.apache.kafka.clients.producer.Producer#close(long, java.util.concurrent.TimeUnit)

Do you have any suggested conf for the timeout here? As for loggers, for servers, for server event loggers?

Or let's introduce proper config next time ?

It makes sense to introduce a new configuration for it. For server's Kafka logger

OK. Will do it in two steps.

introduce a new config kyuubi.backend.server.event.kafka.close.timeout, with default value 60 secs.

pass the config in ServerEventHandlerRegister#createKafkaEventHandler to construct KafkaLoggingEventHandler instance.

60s is too long, it will blocks other shutdown procedures, likely 3~5s

OK, let's make it in 5s by default.

Btw, it's removed for flushing the producer, for the reasons, the closing producer wakes up the internal sender to send pending requests (exact way in flushing), and the flushing method does not accept timeout.

cxzl25 · 2023-05-04T11:34:20Z

kyuubi-events/src/main/scala/org/apache/kyuubi/events/EventLoggerType.scala

@@ -22,5 +22,5 @@ object EventLoggerType extends Enumeration {
  type EventLoggerType = Value

  // TODO: Only SPARK is done now
-  val SPARK, JSON, JDBC, CUSTOM = Value
+  val SPARK, JSON, JDBC, KAFKA, CUSTOM = Value


Should we add enumeration values instead of modifying the order?

KAFKA value is added before CUSTOM here, in order to

follow the same order as in docs

CUSTOM is an extra option, which could be listed behind the options with embedded support

I'm open to your suggestion. WDYT @pan3793 ?

Let's add the new one at the tail, to suppress the potential serializable issue. And the TODO comment is outdated

SGTM. Let me move it to the tail, as you mentioned serializable problem.

pan3793 · 2023-05-08T07:06:38Z

kyuubi-server/pom.xml

+
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka_${scala.binary.version}</artifactId>


I think we are using Testcontainers for integration testing, why should we depend on the Kafka server?

And after second thought, I prefer to use MockProducer rather than launch a real Kafka broker, but the latter is also fine for me.

MockProducer is lightweight to simulate producer-consumer situation, but we need a broker bootstrap server from producer-borker-consumer here. And I don't know how to get a bootstrap server from a MockProducer.

the purpose of the consumer is to verify the events have been successfully delivered to Kafka, we can simply check the buffer of MockProducer instead

pan3793 · 2023-05-08T07:17:36Z

pom.xml

+                <exclusions>
+                    <exclusion>
+                        <groupId>org.slf4j</groupId>
+                        <artifactId>slf4j-api</artifactId>
+                    </exclusion>
+                </exclusions>


unnecessary exclusion

why we should skip this exclusion? kafka-clients is depending on slf4j-api.

We should respect the Maven transitive dependencies in most cases, just excluding the deps we truly don't need, and overriding versions in dependencyManagement(just overriding, without excluding)

But it's not suitable for components which is disastrous for dependencies management, like Hive, Hudi.

SGTM. Removed this exclusion.

kyuubi-server/pom.xml

pan3793 · 2023-05-08T09:31:46Z

NOTICE-binary

@@ -1236,7 +1236,7 @@ This product optionally depends on 'zstd-jni', a zstd-jni Java compression
 and decompression library, which can be obtained at:

  * LICENSE:
-    * license/LICENSE.zstd-jni.txt (Apache License 2.0)


Thanks for fixing this license issue.

🍻 No problem. And we could make this change to a separate PR if you want.

kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala

pan3793 · 2023-05-08T09:34:01Z

kyuubi-events/src/main/scala/org/apache/kyuubi/events/EventLoggerType.scala

@@ -21,6 +21,5 @@ object EventLoggerType extends Enumeration {

  type EventLoggerType = Value

-  // TODO: Only SPARK is done now
-  val SPARK, JSON, JDBC, CUSTOM = Value
+  val SPARK, JSON, JDBC, CUSTOM, KAFKA = Value


all others is supported now?

Seems JDBC is not supported.

I think this TODO can be removed. Not all the event logger types in both servers or the engine. It's a legacy todo hints.

pan3793 · 2023-05-08T09:52:07Z

kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala

+        s" <li>KAFKA: the events serialized in JSON format" +
+        s" and sent to topic of `${SERVER_EVENT_KAFKA_TOPIC.key}`. " +
+        s" Note: For the configs of Kafka producer," +
+        s" please specify them with the prefix: `kyuubi.backend.server.event.kafka.`." +
+        s" For example, kyuubi.backend.server.event.kafka.bootstrap.servers=127.0.0.1:9092" +


Suggested change

s" <li>KAFKA: the events serialized in JSON format" +

s" and sent to topic of `${SERVER_EVENT_KAFKA_TOPIC.key}`. " +

s" Note: For the configs of Kafka producer," +

s" please specify them with the prefix: `kyuubi.backend.server.event.kafka.`." +

s" For example, kyuubi.backend.server.event.kafka.bootstrap.servers=127.0.0.1:9092" +

s" <li>KAFKA: the events will be serialized in JSON format" +

s" and sent to topic of `${SERVER_EVENT_KAFKA_TOPIC.key}`." +

s" Note: For the configs of Kafka producer," +

s" please specify them with the prefix: `kyuubi.backend.server.event.kafka.`." +

s" For example, `kyuubi.backend.server.event.kafka.bootstrap.servers=127.0.0.1:9092`" +

pan3793

Thanks, LGTM, only minor things of docs

pan3793

Thanks, LGTM, only minor things of docs

bowenliang123 · 2023-05-08T10:12:53Z

Thanks, LGTM, only minor things of docs

Thanks for the detailed reviews. Wait for the CI builds and it will be merged. @pan3793 @cxzl25

cxzl25

LGTM

bowenliang123 · 2023-05-08T14:47:00Z

Thanks, merged to master

github-actions bot added kind:build module:common module:events module:server kind:documentation Documentation is a feature! labels Apr 19, 2023

bowenliang123 marked this pull request as ready for review April 19, 2023 11:25

bowenliang123 force-pushed the kafka-logger branch from 8a332c5 to 16cdfb1 Compare April 20, 2023 00:47

bowenliang123 requested review from pan3793 and zhaomin1423 April 20, 2023 01:27