refactor: clean up format pojos in serialized model #3970
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
big-andy-coates
approved these changes
Nov 26, 2019
Comment on lines
238
to
242
| Formats.of( | ||
| outputNode.getKsqlTopic().getKeyFormat(), | ||
| outputNode.getKsqlTopic().getValueFormat(), | ||
| outputNode.getSerdeOptions() | ||
| ), |
There was a problem hiding this comment.
duplicate code, consider pulling out to one call above the if/else.
Comment on lines
80
to
84
| new KsqlTopic( | ||
| createStream.getKafkaTopicName(), | ||
| KeyFormat.of(createStream.getFormats().getKeyFormat(), createStream.getWindowInfo()), | ||
| ValueFormat.of(createStream.getFormats().getValueFormat()) | ||
| ) |
There was a problem hiding this comment.
duplicate code: pull out to a method that takes a CreateSourceCommand and returns KsqlTopic
|
|
||
| @Override | ||
| public int hashCode() { | ||
|
|
| public interface Formats { | ||
| KeyFormat getKeyFormat(); | ||
| public final class Formats { | ||
| // TODO: wrap in a type for type safety |
There was a problem hiding this comment.
outstanding todo...
Adding checkstyle to catch this: confluentinc/common#229
| @JsonProperty(value = "options", required = true) final Set<SerdeOption> options) { | ||
| this.keyFormat = Objects.requireNonNull(keyFormat, "keyFormat"); | ||
| this.valueFormat = Objects.requireNonNull(valueFormat, "valueFormat"); | ||
| this.options = Objects.requireNonNull(options, "options"); |
There was a problem hiding this comment.
Take immutable copy, then you can mark type as immutable.
(Adding test for this: #3978)
This patch cleans up the format pojos in the serialized model, and drops KsqlTopic from the serialized plan. For plan steps that require kafka io (repartitions, changelogs, etc), formats are described by key and value FormatInfo instances, which just contain the serialzation format (json/avro/csv) and required params (schema name, delimiter). The windowing information is now computed either at the source or by the aggregate nodes, and is passed along via the key serde factory when building the streams app. The ddl commands now directly store the information needed about the kafka topic: name, formats, and windowing info for windowed sources.
rodesai
force-pushed
the
clean-up-ksql-topic
branch
from
5bc2109 to
4ebd6bb
Compare
2 tasks
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
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 patch cleans up the format pojos in the serialized model, and
drops KsqlTopic from the serialized plan.
For plan steps that require kafka io (repartitions, changelogs, etc),
formats are described by key and value FormatInfo instances, which
just contain the serialzation format (json/avro/csv) and required
params (schema name, delimiter).
The windowing information is now computed either at the source or
by the aggregate nodes, and is passed along via the key serde
factory when building the streams app.
The ddl commands now directly store the information needed about
the kafka topic: name, formats, and windowing info for windowed
sources.
This is part of an effort to get the plan schemas to the form proposed
here: #3969