fix: added special handling for forwarded pull query request #4597
Conversation
| @@ -51,8 +54,12 @@ | |||
| @Override | |||
| public Optional<ConfigItem> resolve(final String propertyName, final boolean strict) { | |||
| if (propertyName.startsWith(KSQL_CONFIG_PROPERTY_PREFIX) | |||
| && !propertyName.startsWith(KSQL_STREAMS_PREFIX)) { | |||
| && !propertyName.startsWith(KSQL_STREAMS_PREFIX) | |||
There was a problem hiding this comment.
should we rework this block.. Seems like we just want to match a propertyName to three prefixes and call three different methods?
There was a problem hiding this comment.
Yeah but KsqlConfig prefix and and streams prefix both start with ksql. But I simplified the other part
ksql-common/src/main/java/io/confluent/ksql/util/KsqlRequestConfig.java
Outdated
Show resolved
Hide resolved
ksql-engine/src/main/java/io/confluent/ksql/engine/KsqlEngine.java
Outdated
Show resolved
Hide resolved
ksql-engine/src/main/java/io/confluent/ksql/statement/ConfiguredStatement.java
Outdated
Show resolved
Hide resolved
ksql-engine/src/main/java/io/confluent/ksql/statement/ConfiguredStatement.java
Outdated
Show resolved
Hide resolved
ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/execution/PullQueryExecutor.java
Outdated
Show resolved
Hide resolved
ksql-rest-app/src/main/java/io/confluent/ksql/rest/server/execution/PullQueryExecutor.java
Outdated
Show resolved
Hide resolved
| @@ -317,6 +317,63 @@ TableRowsEntity queryRowsLocally( | |||
| ); | |||
| } | |||
|
|
|||
| @VisibleForTesting | |||
| TableRowsEntity forwardTo( | |||
There was a problem hiding this comment.
since this got moved, it was hard to see what changed. but seems like you are just adding the flag.
There was a problem hiding this comment.
Yes, just added the flag
...-streams/src/main/java/io/confluent/ksql/execution/streams/materialization/ks/KsLocator.java
Show resolved
Hide resolved
| List<KsqlHostInfo> allHosts = null; | ||
| if (routingOptions.skipForwardRequest()) { | ||
| LOG.debug("Before filtering: Local host {} ", localHost); | ||
| allHosts = ImmutableList.of(new KsqlHostInfo(localHost.getHost(), localHost.getPort())); |
There was a problem hiding this comment.
I assume this will make node.isLocal() return true always
fixed tests minor change applied vinoth's comments
|
Looks like this introduced a breaking api change see #4674. We need to see what testing we need to catch this kind of stuff early. |
|
On the testing front, I think what we need is something that can generate a swagger/open-api spec file. Then, at build/test time we should validate that any changes to the spec are backwards compatible, and require the dev to commit a new spec if they've made any compatible changes. swagger has some libraries for generating a spec json from jax annotations: https://github.com/swagger-api/swagger-core/wiki/Swagger-Core-Jersey-2.X-Project-Setup-1.5 There are also some tools out there for computing diffs between specs to validate compatibility (e.g. https://github.com/civisanalytics/swagger-diff) |
|
Nice.. Thinking bit more, if we plan to continue supporting the REST API, then may be even writing some tests that do plain http ala curl, vs using the KsqlClient (subclasses) would help catch glaring breaks like this one.. On swagger, it looks interesting. Would it be compatible with vert.x too? makes sense to invest towards that direction, given thats the way of the future |
|
cc @purplefox about swagger. Filed #4684 to track. |
Description
Fixes #4506 and #4413
Before this change we had no way to differentiate between a request that originated from a client and a request internally forwarded to another ksql host due to HA. This resulted in both requests being handled the same way which means for both the logic of filtering and deciding where to route to and actually routing was applied. This could lead to infinite loops of routing between standbys if their lags fluctuated.
With this change, if a request is the result of internal forwarding, the forwarded request is either served locally or not at all and the pull query fails. The only filter applied is the max lag filter of the localhost.
Most notable changes:
Besides changes in
PullQueryExecutorandKsLocator, I created a new config file for internal properties set as part of a pull query request (KsqlRequestConfig) and I also added a map to theKsqlResquestfor these properties.Testing done
Added the properties map to
KsqlRequestTestand verifiedPullQueryRoutingFunctionalTestworks as expected.