Add Protobuf 'any' type support to Kafka connector #17394

adamjshook · 2023-05-08T20:02:58Z

Description

This adds support for the any Protobuf data type, converting the message type to JSON. The default implementation uses the ProtobufUtils mechanism for parsing the Descriptor. There is a separate implementation that uses Confluent's ProtobufSchemaProvider which is used instead when the ConfluentModule is enabled and uses caching when accessing the types.

This PR needs user-facing documentation which will be added as another commit.

Additional context and related issues

The Protobuf any type contains two fields, one type_url field and another being any serialized Protobuf message. The type_url is expected to be a URL that resolves to a serialized descriptor of the message. This change uses a java.net.URL of the type_url to open a stream and read the bytes of the descriptor which is then converted into a Descriptor object. We then pass this to a JSON printer to serialize the message. Afterwards, we have to sort all of the keys, including nested keys, as the Trino JSON type expects all keys to be ordered.

This requires users to ensure that the URL is resolvable from the Coordinator and Worker hosts, either by being a locally installed file:// on each machine or a remote https:// service.

Expands on #16836 which added oneof support

Release notes

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

# Kafka Connector
* Added support for reading Protobuf messages containing the `Any` Protobuf type. This is disabled by default and can be enabled by setting `kafka.protobuf-any-support-enabled` to `true`. Please see docs for more details.

hashhar

lgtm % cache auto-refresh behaviour

hashhar · 2023-05-16T08:35:24Z

lib/trino-record-decoder/src/main/java/io/trino/decoder/protobuf/ProtobufColumnDecoder.java

@@ -60,6 +68,8 @@

 public class ProtobufColumnDecoder
 {
+    private static final ObjectMapper mapper = JsonMapper.builder().configure(ORDER_MAP_ENTRIES_BY_KEYS, true).build();


if the ordering is relevant please add a commmment why. copy the one from sorted method in this class?

hashhar · 2023-05-19T11:51:09Z

...afka/src/main/java/io/trino/plugin/kafka/schema/confluent/ConfluentSchemaRegistryConfig.java

@@ -40,6 +41,8 @@
    private int confluentSchemaRegistryClientCacheSize = 1000;
    private EmptyFieldStrategy emptyFieldStrategy = IGNORE;
    private Duration confluentSubjectsCacheRefreshInterval = new Duration(1, SECONDS);
+    private int confluentDescriptorCacheSize = 1000;
+    private Duration confluentDescriptorCacheRefreshInterval = new Duration(5, MINUTES);


hmmm, IMO the cache shouldn't refresh automatically by default

Updated to remove a configurable cache and instead use a non-evictable cache with a max size of 1000.

adamjshook · 2023-05-25T12:42:44Z

@mosabua Could you please review the documentation for this type support? Thank you!

mosabua

Docs need some clarification and refinement. Great start and thanks for not ignoring the docs @adamjshook

mosabua · 2023-05-29T18:46:41Z

docs/src/main/sphinx/connector/kafka.rst

@@ -88,6 +88,7 @@ Property name                                              Description
 ``kafka.hide-internal-columns``                            Controls whether internal columns are part of the table schema or not.
 ``kafka.internal-column-prefix``                           Prefix for internal columns, defaults to ``_``
 ``kafka.messages-per-split``                               Number of messages that are processed by each Trino split; defaults to ``100000``.
+``kafka.protobuf-any-support-enabled``                     True to enable support for encoding Protobuf ``any`` types to JSON, defaults to ``false``.


How about

Enable support for encoding Protobuf any types to JSON by setting to the property to true, defaults to false.

mosabua · 2023-05-29T18:46:52Z

docs/src/main/sphinx/connector/kafka.rst

 ===================================== =======================================

+any
+++++


same length as title

mosabua · 2023-05-29T18:58:07Z

docs/src/main/sphinx/connector/kafka.rst

+
+Protobuf schemas containing `google.protobuf.Any <https://protobuf.dev/programming-guides/proto3/#any>`_ ``
+fields are mapped to a ``JSON`` field in Trino when
+``kafka.protobuf-any-support-enabled`` is set to ``true``. Any message types


This reads weird .. how about

Message types with an Any field contain an arbitrary ...

docs/src/main/sphinx/connector/kafka.rst

adamjshook · 2023-06-02T18:09:58Z

@mosabua I've made updates to the documentation, please give it another review and let me know of any changes. Thank you!

mosabua

One last nit, but essentially the docs are good to go.

mosabua · 2023-06-05T04:39:02Z

docs/src/main/sphinx/connector/kafka.rst

+        google.protobuf.Any any_message = 1;
+    }
+
+The corresponding Trino column will be named ``any_message`` of type ``JSON``


Suggested change

The corresponding Trino column will be named ``any_message`` of type ``JSON``

The corresponding Trino column is named ``any_message`` of type ``JSON``

adamjshook · 2023-06-16T13:56:00Z

@hashhar I've rebased against master to pick up the changes from #17858.

hashhar

LGTM % some docs comments

hashhar · 2023-06-21T09:56:49Z

docs/src/main/sphinx/connector/kafka.rst

@@ -88,6 +88,8 @@ Property name                                              Description
 ``kafka.hide-internal-columns``                            Controls whether internal columns are part of the table schema or not.
 ``kafka.internal-column-prefix``                           Prefix for internal columns, defaults to ``_``
 ``kafka.messages-per-split``                               Number of messages that are processed by each Trino split; defaults to ``100000``.
+``kafka.protobuf-any-support-enabled``                     Enable support for encoding Protobuf any types to JSON by setting to the property to ``true``,


Suggested change

``kafka.protobuf-any-support-enabled`` Enable support for encoding Protobuf any types to JSON by setting to the property to ``true``,

``kafka.protobuf-any-support-enabled`` Enable support for encoding Protobuf ``any`` types to ``JSON`` by setting to the property to ``true``,

hashhar · 2023-06-21T10:01:12Z

docs/src/main/sphinx/connector/kafka.rst

+
+Message types with an `Any <https://protobuf.dev/programming-guides/proto3/#any>`_
+field contain an arbitrary serialized message as bytes and a type URL to resolve
+that message's type, typically with a scheme of ``file://`` or ``http://``.


are schemes other than file://, http:// or https:// supported? If not let's word this more explicitly to say exactly what schemes are supported.

No, I've changed the docs to be more clear.

hashhar · 2023-07-03T12:50:54Z

Rebased to resolve conflicts. Will merge after CI.

cla-bot bot added the cla-signed label May 8, 2023

adamjshook requested review from hashhar and wendigo May 8, 2023 20:04

adamjshook force-pushed the adamjshook/protobuf-any branch from 1157e13 to b38f960 Compare May 9, 2023 23:04

hashhar reviewed May 19, 2023

View reviewed changes

adamjshook force-pushed the adamjshook/protobuf-any branch 2 times, most recently from cb8a552 to b8e19cf Compare May 25, 2023 12:39

adamjshook added the needs-docs This pull request requires changes to the documentation label May 25, 2023

adamjshook requested a review from hashhar May 25, 2023 12:40

adamjshook requested a review from mosabua May 25, 2023 12:42

github-actions bot added the docs label May 25, 2023

mosabua requested changes May 29, 2023

View reviewed changes

adamjshook force-pushed the adamjshook/protobuf-any branch from b8e19cf to 33d0809 Compare June 2, 2023 16:31

adamjshook requested a review from mosabua June 2, 2023 18:02

mosabua approved these changes Jun 5, 2023

View reviewed changes

adamjshook force-pushed the adamjshook/protobuf-any branch from 33d0809 to a75a191 Compare June 5, 2023 12:55

adamjshook force-pushed the adamjshook/protobuf-any branch from a75a191 to 9c94e5e Compare June 16, 2023 13:54

hashhar approved these changes Jun 21, 2023

View reviewed changes

adamjshook force-pushed the adamjshook/protobuf-any branch from 9c94e5e to 30577b2 Compare June 26, 2023 15:22

adamjshook requested a review from hashhar June 26, 2023 15:22

adamjshook force-pushed the adamjshook/protobuf-any branch from 30577b2 to 47be527 Compare June 26, 2023 17:21

adamjshook added 2 commits July 3, 2023 18:20

Add Protobuf 'any' type support to Kafka connector

5834b8c

Document Protobuf Any support in Kafka connector

29a2d3b

hashhar force-pushed the adamjshook/protobuf-any branch from 47be527 to 29a2d3b Compare July 3, 2023 12:50

hashhar approved these changes Jul 3, 2023

View reviewed changes

hashhar merged commit ee47797 into trinodb:master Jul 3, 2023

github-actions bot added this to the 421 milestone Jul 3, 2023

colebow mentioned this pull request Jul 5, 2023

Add Trino 421 release notes #18075

Merged

	The corresponding Trino column will be named ``any_message`` of type ``JSON``
	The corresponding Trino column is named ``any_message`` of type ``JSON``

	``kafka.protobuf-any-support-enabled`` Enable support for encoding Protobuf any types to JSON by setting to the property to ``true``,
	``kafka.protobuf-any-support-enabled`` Enable support for encoding Protobuf ``any`` types to ``JSON`` by setting to the property to ``true``,

Add Protobuf 'any' type support to Kafka connector #17394

Add Protobuf 'any' type support to Kafka connector #17394

Uh oh!

Conversation

adamjshook commented May 8, 2023 • edited by hashhar 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!

hashhar 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!

adamjshook commented May 25, 2023

Uh oh!

mosabua 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!

Uh oh!

adamjshook commented Jun 2, 2023

Uh oh!

mosabua left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

adamjshook commented Jun 16, 2023

Uh oh!

hashhar 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!

hashhar commented Jul 3, 2023

Uh oh!

Uh oh!

adamjshook commented May 8, 2023 •

edited by hashhar

Loading

hashhar left a comment •

edited

Loading