Add Kafka Transport support in Trino Open Lineage Plugin by alprusty · Pull Request #22998 · trinodb/trino

alprusty · 2024-08-09T23:40:32Z

Description

Currently Trino open lineage plugin supports Console and Http transport. This change is adding a Kafka transport support.

Additional context and related issues

Fixes #21599

Release notes

( ) This is not user-visible or is docs only, and no release notes are required.
( ) Release notes are required. Please propose a release note for me.
( ) Release notes are required, with the following suggested text:

# General
* Add support for Kafka transport in Trino Open lineage Plugin (https://github.com/trinodb/trino/issues/21599)

sopel39

lgtm % testing comment

sopel39 · 2024-08-12T11:28:14Z

...age/src/main/java/io/trino/plugin/openlineage/transport/kafka/OpenLineageKafkaTransport.java

+
+import java.util.Properties;
+
+public class OpenLineageKafkaTransport


Could we add test like TestOpenLineageEventListenerMarquezIntegration.java but with Kafka?

See also #22888 for similar test with Kafka cc @marton-bod

+1 for integration tests with kafka

Praveen2112 · 2024-08-12T11:43:43Z

plugin/trino-openlineage/pom.xml

+                    <failOnWarning>false</failOnWarning>
+                    <ignoredUnusedDeclaredDependencies>
+                        <!-- kafka-clients is needed on class path for KafkaTransport to work in openlineage-java lib -->
+                        <ignoredUnusedDeclaredDependency>kafka-clients:::</ignoredUnusedDeclaredDependency>


Can we use them as a runtime dependency explicitly ?

Praveen2112 · 2024-08-12T11:45:21Z

...age/src/main/java/io/trino/plugin/openlineage/transport/kafka/OpenLineageKafkaTransport.java

+    private final String acks = "acks";
+    private final String acksDefault = "all";
+    private final String keySerializer = "key.serializer";
+    private final String keySerializerDefault = "org.apache.kafka.common.serialization.StringSerializer";
+    private final String valueSerializer = "value.serializer";
+    private final String valueSerializerDefault = "org.apache.kafka.common.serialization.StringSerializer";


How about having them as a static final variables ?

Yup. Either static final variables or configurable via OpenLineageClientKafkaTransportConfig if (some) they need to be changed (prob not)

I think these all producer configuration options should be configurables viaOpenLineageClientKafkaTransportConfig with said defaults. these ones are also required often:

security.protocol
ssl.keystore.location
ssl.keystore.password
ssl.truststore.location
ssl.truststore.password

For the Config key we could use something like

kafkaProperties.put(BOOTSTRAP_SERVERS_CONFIG, brokerEndpoints); kafkaProperties.put(ACKS_CONFIG, acksDefault); kafkaProperties.put(KEY_SERIALIZER_CLASS_CONFIG, keySerializerDefault); kafkaProperties.put(VALUE_SERIALIZER_CLASS_CONFIG, valueSerializerDefault);`

In case of custom protocol - We might need to capture them as a seperate config

Like KafkaSslConfig in Trino

yeah but i would really avoid keeping defaults hardcoded without a way for a user to change, KafkaTransportConfig would be more flexible

mgorsk1

few comments + can you please update docs (event-listeners-openlineage.md)?

mgorsk1 · 2024-08-12T11:54:25Z

...age/src/main/java/io/trino/plugin/openlineage/transport/kafka/OpenLineageKafkaTransport.java

+    private final String acks = "acks";
+    private final String acksDefault = "all";
+    private final String keySerializer = "key.serializer";
+    private final String keySerializerDefault = "org.apache.kafka.common.serialization.StringSerializer";
+    private final String valueSerializer = "value.serializer";
+    private final String valueSerializerDefault = "org.apache.kafka.common.serialization.StringSerializer";


I think these all producer configuration options should be configurables viaOpenLineageClientKafkaTransportConfig with said defaults. these ones are also required often:

security.protocol
ssl.keystore.location
ssl.keystore.password
ssl.truststore.location
ssl.truststore.password

mgorsk1 · 2024-08-12T11:56:17Z

...ain/java/io/trino/plugin/openlineage/config/kafka/OpenLineageClientKafkaTransportConfig.java

+        return this;
+    }
+
+    public String getMessageKey()


shouldn't we have default value for this as there is no NotNull decorator?

mgorsk1 · 2024-08-12T11:56:50Z

...age/src/main/java/io/trino/plugin/openlineage/transport/kafka/OpenLineageKafkaTransport.java

+
+import java.util.Properties;
+
+public class OpenLineageKafkaTransport


+1 for integration tests with kafka

mgorsk1 · 2024-08-12T12:04:22Z

...ain/java/io/trino/plugin/openlineage/config/kafka/OpenLineageClientKafkaTransportConfig.java

+import io.airlift.configuration.ConfigDescription;
+import jakarta.validation.constraints.NotNull;
+
+public class OpenLineageClientKafkaTransportConfig


can you add corresponding TestOpenLineageClientKafkaTransportConfig to tests?

mgorsk1 · 2024-08-12T12:06:48Z

...ain/java/io/trino/plugin/openlineage/config/kafka/OpenLineageClientKafkaTransportConfig.java

+
+    @Config("openlineage-event-listener.transport.kafka.topic")
+    @ConfigDescription("String specifying the topic to which events will be sent")
+    public OpenLineageClientKafkaTransportConfig setTopicName(String topicName)


if the config is openlineage-event-listener.transport.kafka.topic can we unify this and use topic here instead of topicName?

ebyhr · 2024-08-15T07:36:22Z

plugin/trino-openlineage/pom.xml

            <artifactId>jakarta.validation-api</artifactId>
        </dependency>

+        <dependency>


Add Kafka as one of the OL transport types

Please avoid using abbreviations.

ebyhr · 2024-08-15T07:37:12Z

...ain/java/io/trino/plugin/openlineage/config/kafka/OpenLineageClientKafkaTransportConfig.java

+    }
+
+    @NotNull
+    public String getBrokerEndpoints()


Please replace getter/setter orders for messageKey & brokerEndpoints for consistency with fields.

ebyhr · 2024-08-15T07:38:45Z

...ain/java/io/trino/plugin/openlineage/config/kafka/OpenLineageClientKafkaTransportConfig.java

+        return topicName;
+    }
+
+    @Config("openlineage-event-listener.transport.kafka.topic")


I would recommend renaming transport.kafka to kafka-transport. Same for others.

ebyhr · 2024-08-15T07:43:31Z

...ain/java/io/trino/plugin/openlineage/config/kafka/OpenLineageClientKafkaTransportConfig.java

@@ -0,0 +1,66 @@
+/*


Please add a new test to TestOpenLineagePlugin with Kafka transport.

ebyhr · 2024-08-15T07:43:42Z

plugin/trino-openlineage/src/main/java/io/trino/plugin/openlineage/OpenLineageTransport.java

    CONSOLE,
    HTTP,
-    /**/
+    KAFKA


Suggested change

KAFKA

KAFKA,

/**/

ebyhr · 2024-08-15T07:45:23Z

...ain/java/io/trino/plugin/openlineage/config/kafka/OpenLineageClientKafkaTransportConfig.java

+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.trino.plugin.openlineage.config.kafka;


Please use io.trino.plugin.openlineage.transport.kafka package instead. We usually don't add a new package just for config classes.

github-actions · 2024-10-04T17:14:52Z

This pull request has gone a while without any activity. Tagging the Trino developer relations team: @bitsondatadev @colebow @mosabua

mosabua · 2024-10-04T20:26:53Z

👋 @alprusty .. can you rebase and address the comments from @ebyhr @mgorsk1 and @sopel39 ?

alprusty · 2024-10-04T23:40:32Z

Thanks @mosabua for following up here.
Yes, will rebase and take care of review comments in a follow up commit.

github-actions · 2024-10-29T17:03:09Z

This pull request has gone a while without any activity. Tagging the Trino developer relations team: @bitsondatadev @colebow @mosabua

github-actions · 2024-11-20T17:03:05Z

Closing this pull request, as it has been stale for six weeks. Feel free to re-open at any time.

mosabua · 2024-11-20T17:23:16Z

I am reopening and addig stale-ignore label under the assumption that you will continue this work @alprusty

github-actions · 2025-03-19T17:03:41Z

This pull request has gone a while without any activity. Tagging for triage help: @mosabua

ebyhr · 2025-03-28T00:52:40Z

Closing as a stale PR. Please feel free to reopen if you continue the work.

Add Kafka as one of the OL transport types

fae216f

cla-bot bot added the cla-signed label Aug 9, 2024

alprusty requested review from Praveen2112, mgorsk1 and sopel39 August 10, 2024 00:11

sopel39 reviewed Aug 12, 2024

View reviewed changes

Praveen2112 reviewed Aug 12, 2024

View reviewed changes

mgorsk1 suggested changes Aug 12, 2024

View reviewed changes

ebyhr reviewed Aug 15, 2024

View reviewed changes

ebyhr assigned alprusty Sep 1, 2024

github-actions bot added the stale label Oct 4, 2024

github-actions bot removed the stale label Oct 7, 2024

github-actions bot added the stale label Oct 29, 2024

github-actions bot closed this Nov 20, 2024

mosabua reopened this Nov 20, 2024

mosabua added stale-ignore Use this label on PRs that should be ignored by the stale bot so they are not flagged or closed. and removed stale labels Nov 20, 2024

ebyhr removed the stale-ignore Use this label on PRs that should be ignored by the stale bot so they are not flagged or closed. label Feb 26, 2025

github-actions bot added the stale label Mar 19, 2025

ebyhr closed this Mar 28, 2025

kekwan mentioned this pull request Jan 3, 2026

Enable Kafka Transport for OpenLineage Event Listener #27830

Open


		import java.util.Properties;

		public class OpenLineageKafkaTransport

Conversation

alprusty commented Aug 9, 2024 • edited by ebyhr Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Additional context and related issues

Release notes

Uh oh!

sopel39 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

sopel39 Aug 12, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mgorsk1 left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ebyhr Aug 15, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ebyhr Aug 15, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ebyhr Aug 15, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

github-actions bot commented Oct 4, 2024

Uh oh!

mosabua commented Oct 4, 2024

Uh oh!

alprusty commented Oct 4, 2024

Uh oh!

github-actions bot commented Oct 29, 2024

Uh oh!

github-actions bot commented Nov 20, 2024

Uh oh!

mosabua commented Nov 20, 2024

Uh oh!

github-actions bot commented Mar 19, 2025

Uh oh!

ebyhr commented Mar 28, 2025

Uh oh!

alprusty commented Aug 9, 2024 •

edited by ebyhr

Loading

sopel39 Aug 12, 2024 •

edited

Loading

mgorsk1 left a comment •

edited

Loading

ebyhr Aug 15, 2024 •

edited

Loading

ebyhr Aug 15, 2024 •

edited

Loading

ebyhr Aug 15, 2024 •

edited

Loading