KSQL-398: integration test covering quickstart scenarios #431
Conversation
|
@confluentinc It looks like @apurvam just signed our Contributor License Agreement. 👍 Always at your service, clabot |
|
retest this please |
| @@ -112,6 +112,10 @@ public void sql(String sql) throws Exception { | |||
| return ksqlEngine.getLiveQueries(); | |||
| } | |||
|
|
|||
| public KsqlEngine getKsqlEngine() { | |||
There was a problem hiding this comment.
Do we want to expose this? i.e., it seems it is only required for the purpose of the test.
There was a problem hiding this comment.
As I said below, we need a reference to the ksqlEngine to submit non-persistent queries. The integration test is in a separate package, so it can't be package-private. Open to suggestions for how to resolve this. One option is to enable the ksqlContext.sql method to run non-persistent queries. But I am not sure if that makes sense from an API perspective.
There was a problem hiding this comment.
I think we need another way where we don't expose the KsqlEngine from an API PoV it is a big increase in the surface area
There was a problem hiding this comment.
@bluemonk3y @hjafarpour thoughts on this? i.e, it seems that the KsqlContext is pretty limited. Is this ok? Should we make it a bit more useable, so we can easily do what @apurvam is trying to achieve without having to get a handle to the ksqlEngine?
There was a problem hiding this comment.
I agree, KsqlContext should not expose the engine.
@apurvam maybe you can use KsqlEngine directly in your test.
There was a problem hiding this comment.
I updated the code to use KSQLEngine directly.
| String query = String.format("SELECT * from %s;", userTable); | ||
| log.debug("Sending query: {}", query); | ||
|
|
||
| List<QueryMetadata> queries = ksqlContext.getKsqlEngine().buildMultipleQueries(false, query, Collections.emptyMap()); |
There was a problem hiding this comment.
Can we use ksqlContext.sql(...) here and then call ksqlContext.getRunningQueries()?
Same in the other tests
There was a problem hiding this comment.
We can't use ksqlContext.sql directly because it has an explicit check that the queries submitted through that method are persistent queries. These are not persistent queries, so they need to be submitted directly to the ksqlEngine.
There was a problem hiding this comment.
Seems a bit of a shame. Means that KsqlContext is extremely limiting! I know not for this PR, just saying
| BlockingQueue<KeyValue<String, GenericRow>> rowQueue = queryMetadata.getRowQueue(); | ||
|
|
||
| Set<String> actualUsers = new HashSet<>(); | ||
| Set<String> expectedUsers = new HashSet<>(Arrays.asList("USER_0", "USER_1", "USER_2", "USER_3", "USER_4")); |
| List<String> actualPages = new ArrayList<>(); | ||
| List<String> actualUsers = new ArrayList<>(); | ||
|
|
||
| while (results.size() < 3) { |
There was a problem hiding this comment.
Do we need a timeout on this loop? Or maybe use TestUtils#waitForCondition? i.e., is there a chance this will loop forever?
There was a problem hiding this comment.
Yea. I was thinking about that. It's better to just fail out if we don't get anything after a minute or something, rather than having the test hang.
| List<String> actualPages = new ArrayList<>(); | ||
| List<String> expectedPages = Arrays.asList("PAGE_5"); | ||
|
|
||
| while (actualPages.size() < 1) { |
| replicatonFactor) { | ||
| log.info("Creating topic '{}'", topic); | ||
|
|
||
| public void createTopic(final String topic, final int numPartitions, final short replicatonFactor) { |
There was a problem hiding this comment.
This method overrides a the same method from the interface. It needs @Override annotation.
There was a problem hiding this comment.
Ah. That is a rebase mistake. Will clean it up.
|
retest this please |
…. IT fails right now.
apurvam
force-pushed
the
KSQL-398-integration-test-covering-quickstart-scenarios
branch
from
25403c6
to
c63600b
Compare
|
I'm still waiting for one more LGTM before merging, as per our agreed conventions. |
apurvam
deleted the
KSQL-398-integration-test-covering-quickstart-scenarios
branch
This patch covers all the queries in the quickstart, including the 'plain' select statements.
I have noticed that the join between the pageviews stream and the users table sometimes results in 'null' values for the columns in the users table. @hjafarpour thinks that this is because the rocksdb store isn't ready when the join is executed, resulting in an empty user table and null values for the corresponding fields in the join.