Kafka protobuf consumer support #4375
Merged
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This adds support for kafka serdes-protobuf deserialization via KafkaTools#protobufSpec. There is an extensive amount of unit testing, both in the form of "how does this message structure gets parsed into column types" and "how do schema changes get adapted into the same table". While this feature is primarily targetted for use via KafkaTools, the underlying extensions-protobuf library is generalized to work in other cases as well. For example, it would work just as well at parsing protobuf messages via gRPC or websockets (at which point, it could easily be adapted into a blink table). The implementation is underpinned by a set a functional interfaces for extracting primitives or Objects from a generic type; in the case of protobuf, the generic type is com.google.protobuf.Message. Coupled with functional composition, these interfaces provide useful primitives for fluently expressing and executing parsing expressions in a reusable way. See https://docs.confluent.io/platform/current/schema-registry/fundamentals/serdes-develop/serdes-protobuf.html Fixes deephaven#3832
devinrsmith
added
kafka
DocumentationNeeded
ReleaseNotesNeeded
devinrsmith
requested review from
nbauernfeind,
niloc132,
rcaudy,
chipkent and
jmao-denver
as code owners
|
Note: a large part of the +20k diff is checked-in protobuf test code - see notes in TESTING.md for why I had to check it in. |
| } | ||
|
|
||
| private Optional<ProtobufFunctions> wellKnown() { | ||
| // todo: eventually have support for parsing specific fields in specific ways |
There was a problem hiding this comment.
do you want to track this with an issue?
There was a problem hiding this comment.
I'm confused about this. Are we recursively expanding nested fields?
extensions/protobuf/src/main/java/io/deephaven/protobuf/ProtobufDescriptorParserImpl.java
Outdated
Show resolved
Hide resolved
rcaudy
reviewed
Sep 5, 2023
extensions/protobuf/src/main/java/io/deephaven/protobuf/ProtobufDescriptorParserOptions.java
Outdated
Show resolved
Hide resolved
| * </tr> | ||
| * </table> | ||
| * | ||
| * ^1 Unsigned 32-bit and 64-bit integers are represented using their signed counterpart, with the top bit being |
There was a problem hiding this comment.
This is not what we typically do.
There was a problem hiding this comment.
I'll add documentation to java.lang.Integer#toUnsignedLong, which I feel is the appropriate way to handle this from the users' perspective if they want this behavior.
There was a problem hiding this comment.
Also, java.lang.Long#toUnsignedBigInteger exists, but it's private :(
There was a problem hiding this comment.
If you feel very strongly, or as additional feature in the future, we could introduce this behavior as part of FieldOptions...
extensions/protobuf/src/main/java/io/deephaven/functions/ObjectFunction.java
Outdated
Show resolved
Hide resolved
extensions/protobuf/src/main/java/io/deephaven/functions/ObjectFunction.java
Outdated
Show resolved
Hide resolved
extensions/protobuf/src/main/java/io/deephaven/functions/ObjectFunctions.java
Outdated
Show resolved
Hide resolved
extensions/protobuf/src/main/java/io/deephaven/protobuf/ProtobufDescriptorParserImpl.java
Outdated
Show resolved
Hide resolved
extensions/kafka/src/main/java/io/deephaven/kafka/ProtobufImpl.java
Outdated
Show resolved
Hide resolved
extensions/kafka/src/main/java/io/deephaven/kafka/CommonTransform.java
Outdated
Show resolved
Hide resolved
extensions/kafka/src/main/java/io/deephaven/kafka/ProtobufImpl.java
Outdated
Show resolved
Hide resolved
rcaudy
reviewed
Sep 6, 2023
extensions/protobuf/src/main/java/io/deephaven/protobuf/ProtobufFunction.java
Outdated
Show resolved
Hide resolved
extensions/kafka/src/main/java/io/deephaven/kafka/ProtobufImpl.java
Outdated
Show resolved
Hide resolved
extensions/kafka/src/main/java/io/deephaven/kafka/protobuf/ProtobufConsumeOptions.java
Outdated
Show resolved
Hide resolved
extensions/protobuf/src/main/java/io/deephaven/protobuf/MessageParserAdapter.java
Outdated
Show resolved
Hide resolved
extensions/protobuf/src/main/java/io/deephaven/protobuf/MessageParserSingle.java
Outdated
Show resolved
Hide resolved
rcaudy
previously approved these changes
Sep 6, 2023
jmao-denver
reviewed
Sep 7, 2023
jmao-denver
previously approved these changes
Sep 7, 2023
jmao-denver
approved these changes
Sep 7, 2023
rcaudy
approved these changes
Sep 7, 2023
|
Labels indicate documentation is required. Issues for documentation have been opened: How-to: https://github.com/deephaven/deephaven.io/issues/3150 |
Sign up for free
to subscribe to this conversation on GitHub.
Already have an account?
Sign in.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
This adds support for kafka serdes-protobuf deserialization via KafkaTools#protobufSpec.
There is an extensive amount of unit testing, both in the form of "how does this message structure gets parsed into column types" and "how do schema changes get adapted into the same table".
While this feature is primarily targetted for use via KafkaTools, the underlying extensions-protobuf library is generalized to work in other cases as well. For example, it would work just as well at parsing protobuf messages via gRPC or websockets (at which point, it could easily be adapted into a blink table).
The implementation is underpinned by a set a functional interfaces for extracting primitives or Objects from a generic type; in the case of protobuf, the generic type is com.google.protobuf.Message. Coupled with functional composition, these interfaces provide useful primitives for fluently expressing and executing parsing expressions in a reusable way.
See https://docs.confluent.io/platform/current/schema-registry/fundamentals/serdes-develop/serdes-protobuf.html
Fixes #3832