feat: wire in the KS config needed for point queries (MINOR) #3251
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
Point queries need to be able to determine with KSQL node stores the data for a specific key in a state store. Kafka Streams supports this via its interactive queries functionality. For this to work, the KS `application.server` config must be set. This change sees the `application.server` being automatically set by the rest application to the FIRST listener supplied in the standard `listener` config. The `listener` can use automatic port allocation, e.g. if set to `http://localhost:0` then it will auto-allocate a port number. This means the port is not known until the server has been started, which requires the `KsqlConfig` to be injected into certain types once the server is started.
big-andy-coates
commented
Aug 23, 2019
| @@ -115,7 +116,7 @@ | |||
|
|
|||
| public static final String COMMANDS_STREAM_NAME = "KSQL_COMMANDS"; | |||
|
|
|||
| private final KsqlConfig ksqlConfig; | |||
| private final KsqlConfig ksqlConfigNoPort; | |||
There was a problem hiding this comment.
I've renamed this to make it clear it does not contain the port info. My thinking being that people may think twice before using it blindly.
big-andy-coates
commented
Aug 23, 2019
| @@ -198,6 +195,8 @@ public void setupResources(final Configurable<?> config, final KsqlRestConfig ap | |||
| @Override | |||
| public void start() throws Exception { | |||
| super.start(); | |||
| final KsqlConfig ksqlConfigWithPort = buildConfigWithPort(); | |||
| configurables.forEach(c -> c.configure(ksqlConfigWithPort)); | |||
There was a problem hiding this comment.
Once the Jetty server has been started we know the actual ports, so can now build more complete config and configure those components that are lazily configured.
big-andy-coates
commented
Aug 23, 2019
| @@ -109,7 +109,6 @@ public void setup() { | |||
| when(serviceContext.getTopicClient()).thenReturn(kafkaTopicClient); | |||
| when(serviceContext.getSchemaRegistryClient()).thenReturn(schemaRegistryClient); | |||
| clusterTerminator = new ClusterTerminator( | |||
| ksqlConfig, | |||
There was a problem hiding this comment.
removed, as its not used.
agavra
approved these changes
Aug 23, 2019
ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/KsqlRestApplication.java
Show resolved
Hide resolved
ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/resources/KsqlResource.java
Show resolved
Hide resolved
# Conflicts: # ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/KsqlRestApplication.java # ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/resources/KsqlResource.java # ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/resources/streaming/StreamedQueryResource.java
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.
Description
Point queries need to be able to determine with KSQL node stores the data for a specific key in a state store. Kafka Streams supports this via its interactive queries functionality. For this to work, the KS
application.serverconfig must be set.This change sees the
application.serverbeing automatically set by the rest application to the FIRST listener supplied in the standardlistenerconfig.As the
listenersetting can use automatic port allocation, e.g. if set tohttp://localhost:0then it will auto-allocate a port number, we might not know the actual port until the Jetty server is started.This causes problems where components require the
KsqlConfigto be injected at creation time. To get around this I've addedKsqlConfigurable, which components can implement, meaning they get passed their config after creation, and that config includes the required app server setting with the correct port.The following components currently implement
KsqlConfigurable:KsqlResourceStreamedQueryResourceStatementExecutorStrictly speaking, only the last of these currently needs the app server set, as it is the only one the currently starts persistent queries. But I think it's safer, i.e. less likely to cause bugs in the future, if all three are configured with the app server set.
Testing done
Some tests added in this PR. More coming with later point query PRs.
Reviewer checklist