Add topic, partition, and timestamp column kafka publishing support #4771
Conversation
This adds optional features for publishing a kafka record with the topic, partition, and/or timestamp as specified by their respective columns. Additionally, this also adds the ability to easily specify a default partition. Fixes deephaven#4767
devinrsmith
added
feature request
devinrsmith
requested review from
chipkent,
jmao-denver and
rcaudy
as code owners
extensions/kafka/src/main/java/io/deephaven/kafka/KafkaPublishOptions.java
Show resolved
Hide resolved
extensions/kafka/src/main/java/io/deephaven/kafka/KafkaPublishOptions.java
Show resolved
Hide resolved
extensions/kafka/src/main/java/io/deephaven/kafka/KafkaPublishOptions.java
Outdated
Show resolved
Hide resolved
extensions/kafka/src/main/java/io/deephaven/kafka/publish/PublishToKafka.java
Outdated
Show resolved
Hide resolved
extensions/kafka/src/main/java/io/deephaven/kafka/publish/PublishToKafka.java
Outdated
Show resolved
Hide resolved
Adds some helper methods for on `TableDefinition#checkHasColumn`, `TableDefinition#checkHasColumns`, and `TableDefinition#getColumnNameSet`. Additionally, fixes up call sites that were (ab)using `Table#getColumnSourceMap` to simply get the keySet. This invokes a potentially extraneous Table#coalesce which can be avoided in these cases. In support of common scaffolding so #4771 won't need to call `Table#getColumnSource` for validation purposes.
| * | ||
| * @param destDataType the destination data type | ||
| */ | ||
| public final void checkCastTo(Class<?> destDataType) { |
There was a problem hiding this comment.
Usually we just have a cast method that returns the type we like and (maybe) validates internally.
There was a problem hiding this comment.
I think it's a little bit different w/ ColumnDefinition. In other cases cases where cast is used, it is typically done to match types b/c the consumer wants to get data out. In this case, ColumnDefinition itself doesn't hold any data, it is just parameterized to support explicit typing around Class <TYPE> getDataType().
I don't see a practical use for wanting to actually cast a definition. Furthermore, it breaks the contract:
ColumnDefinition<Integer> intColDef = ...;
ColumnDefinition<Number> numColDef = intColDef.cast(Number.class);
Class<Number> numberClass = numColDef.getDataType();
// This will fail
assertEquals(Number.class, numberClass);We do play loose and fast with this in other places; ColumnSource suffers from the same getType problem - although again, ColumnSource#cast is usually used to get data out (not b/c somebody wants getType).
In some more recent code, I've tried to "do the right thing"; io.deephaven.functions.ToObjectFunction#cast
| * @param destDataType the destination data type | ||
| */ | ||
| public final void checkCastTo(Class<?> destDataType) { | ||
| TypeHelper.checkCastTo("[" + name + "]", dataType, destDataType); |
There was a problem hiding this comment.
This context being passed does not mesh well with the exception message as currently composed.
There was a problem hiding this comment.
I think you took back this point.
engine/api/src/main/java/io/deephaven/engine/table/ColumnDefinition.java
Show resolved
Hide resolved
engine/api/src/main/java/io/deephaven/engine/table/TableDefinition.java
Outdated
Show resolved
Hide resolved
engine/api/src/main/java/io/deephaven/engine/table/TableDefinition.java
Outdated
Show resolved
Hide resolved
extensions/kafka/src/main/java/io/deephaven/kafka/KafkaPublishOptions.java
Outdated
Show resolved
Hide resolved
extensions/kafka/src/main/java/io/deephaven/kafka/KafkaPublishOptions.java
Outdated
Show resolved
Hide resolved
extensions/kafka/src/main/java/io/deephaven/kafka/KafkaPublishOptions.java
Outdated
Show resolved
Hide resolved
extensions/kafka/src/main/java/io/deephaven/kafka/KafkaPublishOptions.java
Outdated
Show resolved
Hide resolved
|
Labels indicate documentation is required. Issues for documentation have been opened: How-to: https://github.com/deephaven/deephaven.io/issues/3450 |
This adds optional configuration for publishing a kafka record with the topic, partition, and/or timestamp as specified by their respective columns.
Additionally, this also adds the ability to specify a default partition.
Fixes #4767