Kafka integration #983

TomBebb · 2024-11-08T03:27:25Z

/claim #936

Fixes #936

Description

Integrates with Kakfa, using its crate.
This is toggled on/off via the KAFKA_TOPIC environment variable.
This uses a background polling iterator to loop through messages, and ingesting messages to the stream with the name of the KAFKA_TOPIC.
Covers all possible config options via environment variables.

This PR has:

[ y] been tested to ensure log ingestion and log query works.
[ y] added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
[ y] added documentation for new or modified features or behaviors.

# Conflicts: # Cargo.toml

TomBebb · 2024-11-08T14:43:13Z

Kafka-console-producer:

Data:

de-sh · 2024-11-15T17:25:27Z

server/src/kafka.rs

+    Ok(())
+}
+
+pub async fn setup_integration() -> Result<JoinHandle<()>, KafkaError> {


It should be fine to drop the handle if I am not wrong, why are we returning it?

de-sh · 2024-11-15T17:29:09Z

server/Cargo.toml

@@ -106,6 +106,8 @@ path-clean = "1.0.1"
 prost = "0.13.3"
 prometheus-parse = "0.2.5"
 sha2 = "0.10.8"
+kafka = "0.10.0"


Using https://github.com/fede1024/rust-rdkafka instead of kafka-rust could allow the implementation to be async

de-sh · 2024-11-19T06:57:26Z

@TomBebb can you please document the code a bit more, this will help us understand what we are going to maintain in the long run 😅

The code here is lacking comments, especially regarding the functions and what they are doing as well as the use of good const(The env variables should be named in const as the deepsource action points out) that describes what they are doing. Thank you in advance!

de-sh · 2024-11-19T17:41:02Z

server/src/kafka.rs

+            loop {
+                while let Some(curr) = stream.next().await {
+                    let msg = curr.unwrap();
+                    ingest_message(&stream_name, msg).await.unwrap();
+                }
+            }


while let inside a loop? Also, why the unwrap is not after stream.next() and instead on the returned value when ref says to do otherwise?

de-sh · 2024-11-19T17:42:19Z

.gitignore

@@ -14,4 +14,5 @@ parseable
 parseable_*
 parseable-env-secret
 cache
+.idea


please remove unrelated change

de-sh · 2024-11-19T17:43:14Z

.idea/.gitignore

same as above

de-sh · 2024-11-19T17:44:14Z

server/src/kafka.rs

+    Ok(())
+}
+
+pub async fn setup_integration() -> Result<JoinHandle<()>, KafkaError> {


Suggested change

pub async fn setup_integration() -> Result<JoinHandle<()>, KafkaError> {

pub async fn setup_integration() -> Result<(), KafkaError> {

de-sh · 2024-11-19T17:44:30Z

server/src/kafka.rs

+        log::info!("Setup kafka integration for {stream_name}");
+        create_stream_if_not_exists(&stream_name, &StreamType::UserDefined.to_string()).await?;
+
+        let res = task::spawn(async move {


Suggested change

let res = task::spawn(async move {

task::spawn(async move {

de-sh · 2024-11-19T17:44:43Z

server/src/kafka.rs

+}
+
+pub async fn setup_integration() -> Result<JoinHandle<()>, KafkaError> {
+    let my_res = if let Ok(stream_name) = env::var("KAFKA_TOPIC") {


Suggested change

let my_res = if let Ok(stream_name) = env::var("KAFKA_TOPIC") {

if let Ok(stream_name) = env::var("KAFKA_TOPIC") {

de-sh · 2024-11-19T17:45:04Z

server/src/kafka.rs

+    } else {
+        task::spawn_blocking(|| {})
+    };


Suggested change

} else {

task::spawn_blocking(|| {})

};

}

de-sh · 2024-11-19T17:45:16Z

server/src/kafka.rs

+    } else {
+        task::spawn_blocking(|| {})
+    };
+    Ok(my_res)


Suggested change

Ok(my_res)

Ok(())

parmesant · 2024-11-29T06:40:56Z

Hey @TomBebb !
I wanted to try out your changes but am running into some issues

Steps

setup kafka from binary and start the server
set KAFKA_TOPIC env var
set KAFKA_HOSTS env var

run parseable (local-store) - I can see the logstream get created on the console
start a bootstrap server and send a JSON message - I don't see any debug logs on parseable or any log events on console

close the bootstrap server - I still don't see any logs on console

Did I miss some step?

TomBebb added 23 commits October 21, 2024 20:43

Add kafka dependency and setup basic demo

7a658de

Implement basic env var reading / message reading

dcd496b

Support client_id override

fb94dca

Support int envs to builder mapping

8f1bf00

Support durations

11b8d7d

Handle env parsing error better

7162580

Show formatted errors on environment variable parsing error

ce8d727

Sync kafka messages

1d0905e

Tidy imports

d708659

Fix double-import

2317da9

Fix metadata error typo

3d2f3b1

Add RustRover IDE folder to gitignore

f68b7b1

Merge branch 'main' into kafka-integration

4648715

# Conflicts: # Cargo.toml

Merge branch 'main' into kafka-integration

28e316a

Simplify log push

4299c49

Fix kafka order

8c4dbae

Improve kafka order

ed685c6

Support kafka group/partition envs

27c3f0b

Simplify parseinterror prefix

90a74cc

Support fetch_crc_validation

229bde8

Support kafka offset storage

fc86edc

Support kafka security

1eb536f

Support kafka fallback offset

17bdb25

algora-pbc bot mentioned this pull request Nov 8, 2024

Feat: Add native Kafka integration for Parseable server #936

Open

algora-pbc bot added the 🙋 Bounty claim label Nov 8, 2024

Fix security default value

0ffc027

TomBebb added 2 commits November 10, 2024 22:38

Merge branch 'main' into kafka-integration

e2c04eb

Merge branch 'main' into kafka-integration

65e38ad

nitisht requested a review from nikhilsinhaparseable November 15, 2024 16:35

nitisht requested a review from de-sh November 15, 2024 16:36

de-sh reviewed Nov 15, 2024

View reviewed changes

TomBebb added 2 commits November 18, 2024 17:58

Use rdkafka instead of kafka lib

0d518a2

Use async message ingestion

b94a933

de-sh reviewed Nov 19, 2024

View reviewed changes

server/src/kafka.rs

} else {

task::spawn_blocking(|| {})

};

Ok(my_res)

Copy link

Contributor

de-sh Nov 19, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change

Ok(my_res)

Ok(())

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Kafka integration #983

Kafka integration #983

TomBebb commented Nov 8, 2024

TomBebb commented Nov 8, 2024

de-sh Nov 15, 2024

de-sh Nov 15, 2024

de-sh commented Nov 19, 2024 •

edited

Loading

de-sh Nov 19, 2024

de-sh Nov 19, 2024

de-sh Nov 19, 2024

de-sh Nov 19, 2024

de-sh Nov 19, 2024

de-sh Nov 19, 2024

de-sh Nov 19, 2024

de-sh Nov 19, 2024

parmesant commented Nov 29, 2024

	pub async fn setup_integration() -> Result<JoinHandle<()>, KafkaError> {
	pub async fn setup_integration() -> Result<(), KafkaError> {

	let my_res = if let Ok(stream_name) = env::var("KAFKA_TOPIC") {
	if let Ok(stream_name) = env::var("KAFKA_TOPIC") {

Kafka integration #983

Are you sure you want to change the base?

Kafka integration #983

Conversation

TomBebb commented Nov 8, 2024

Description

TomBebb commented Nov 8, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

de-sh commented Nov 19, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

parmesant commented Nov 29, 2024

de-sh commented Nov 19, 2024 •

edited

Loading