You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by GitBox <gi...@apache.org> on 2022/08/29 03:05:41 UTC
[GitHub] [druid] LakshSingla opened a new pull request, #12992: Add IT for MSQ task engine using the new IT framework
LakshSingla opened a new pull request, #12992:
URL: https://github.com/apache/druid/pull/12992
### Description
This PR aims to add IT for the MSQ task engine using the new IT framework.
Key classes added as part of this PR :
* `org.apache.druid.testsEx.categories.MultiStageQuery` - Category for the ITs that are executed using MSQ
* `org.apache.druid.testsEx.msq.ITMultiStageQuery` - File containing the actual test case
* `org.apache.druid.testing.clients.MsqOverlordResourceTestClient` - Overlord resource client for Msq specific operations
* `org.apache.druid.testing.clients.MsqTestClient`- Query resource client for MSQ calls
* `org.apache.druid.testing.utils.MsqQueryWithResults` - Helper POJO that contains an MSQ query (SQL String) and the results which can be tested easily
* `org.apache.druid.testing.utils.MsqTestQueryHelper` - Helper class containing utils to easily write test cases for MSQ task engine
<hr>
<!-- Check the items by putting "x" in the brackets for the done things. Not all of these items apply to every PR. Remove the items which are not done or not relevant to the PR. None of the items from the checklist below are strictly necessary, but it would be very helpful if you at least self-review the PR. -->
This PR has:
- [ ] been self-reviewed.
- [ ] using the [concurrency checklist](https://github.com/apache/druid/blob/master/dev/code-review/concurrency.md) (Remove this item if the PR doesn't have any relation to concurrency.)
- [ ] added documentation for new or modified features or behaviors.
- [ ] added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
- [ ] added or updated version, license, or notice information in [licenses.yaml](https://github.com/apache/druid/blob/master/dev/license.md)
- [ ] added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
- [ ] added unit tests or modified existing tests to cover new code paths, ensuring the threshold for [code coverage](https://github.com/apache/druid/blob/master/dev/code-review/code-coverage.md) is met.
- [ ] added integration tests.
- [ ] been tested in a test Druid cluster.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] LakshSingla commented on a diff in pull request #12992: Add IT for MSQ task engine using the new IT framework
Posted by GitBox <gi...@apache.org>.
LakshSingla commented on code in PR #12992:
URL: https://github.com/apache/druid/pull/12992#discussion_r968140455
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/MsqTestQueryHelper.java:
##########
@@ -0,0 +1,236 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.utils;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import org.apache.druid.indexer.TaskState;
+import org.apache.druid.indexer.TaskStatusPlus;
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.guava.Yielder;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.msq.indexing.report.MSQResultsReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReportPayload;
+import org.apache.druid.msq.sql.SqlTaskStatus;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.sql.http.SqlQuery;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.MsqOverlordResourceTestClient;
+import org.apache.druid.testing.clients.MsqTestClient;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+/**
+ * Helper class to aid out ITs for MSQ
+ */
+public class MsqTestQueryHelper extends AbstractTestQueryHelper<MsqQueryWithResults>
Review Comment:
LMK if the class design and the Javadoc is more legible post the changes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] LakshSingla commented on a diff in pull request #12992: Add IT for MSQ task engine using the new IT framework
Posted by GitBox <gi...@apache.org>.
LakshSingla commented on code in PR #12992:
URL: https://github.com/apache/druid/pull/12992#discussion_r973825160
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/QueryResultVerifier.java:
##########
@@ -19,13 +19,19 @@
package org.apache.druid.testing.utils;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.logger.Logger;
+
import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
public class QueryResultVerifier
{
- public static boolean compareResults(
+ private static final Logger LOG = new Logger(QueryResultVerifier.class);
+
+ public static Optional<String> compareResults(
Review Comment:
I would defer against the suggestion of `QueryResultVerifier.assertEqual(...);` because the helper is not going to do assertions based on the result. This is because the same method is utilized in `ITQueryRetryTestOnMissingSegments.java` which depending on a flag fails or drops the mismatch in the rows. Therefore one way would be to create a separate method that helps perform the Assertions. But since this is not required in several places, I haven't created a such method.
Good point regarding the test case in the planner. That would certainly be a cleaner and an easier approach to add tests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] LakshSingla commented on a diff in pull request #12992: Add IT for MSQ task engine using the new IT framework
Posted by GitBox <gi...@apache.org>.
LakshSingla commented on code in PR #12992:
URL: https://github.com/apache/druid/pull/12992#discussion_r973838702
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/AbstractTestQueryHelper.java:
##########
@@ -146,24 +147,29 @@ private void testQueries(String url, List<QueryResultType> queries) throws Excep
for (QueryResultType queryWithResult : queries) {
LOG.info("Running Query %s", queryWithResult.getQuery());
List<Map<String, Object>> result = queryClient.query(url, queryWithResult.getQuery());
- if (!QueryResultVerifier.compareResults(result, queryWithResult.getExpectedResults(),
- queryWithResult.getFieldsToTest()
- )) {
+ Optional<String> resultsComparison = QueryResultVerifier.compareResults(result, queryWithResult.getExpectedResults(),
+ queryWithResult.getFieldsToTest());
+ if (resultsComparison.isPresent()) {
LOG.error(
"Failed while executing query %s \n expectedResults: %s \n actualResults : %s",
queryWithResult.getQuery(),
jsonMapper.writeValueAsString(queryWithResult.getExpectedResults()),
jsonMapper.writeValueAsString(result)
);
- failed = true;
+ Assert.fail(StringUtils.format(
+ "Results mismatch while executing the query %s.\n"
+ + "Expected results: %s\n"
+ + "Actual results: %s\n"
+ + "Mismatch error: %s\n",
+ queryWithResult.getQuery(),
+ jsonMapper.writeValueAsString(queryWithResult.getExpectedResults()),
+ jsonMapper.writeValueAsString(result),
+ resultsComparison.get()
+ ));
Review Comment:
Removed the expected and actual results part because that would be the two large JSON blobs that you mentioned here. The row number should be informed to the user in the test case. As far as the formatting of the results is concerned, I think it might be difficult to do so once the file has been deserialized into an object. One way would be to serialize the expected result the opposite way the deserialization has happened.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] abhishekagarwal87 merged pull request #12992: Add IT for MSQ task engine using the new IT framework
Posted by GitBox <gi...@apache.org>.
abhishekagarwal87 merged PR #12992:
URL: https://github.com/apache/druid/pull/12992
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] kfaraz commented on a diff in pull request #12992: Add IT for MSQ task engine using the new IT framework
Posted by GitBox <gi...@apache.org>.
kfaraz commented on code in PR #12992:
URL: https://github.com/apache/druid/pull/12992#discussion_r972796940
##########
integration-tests/src/main/java/org/apache/druid/testing/clients/msq/MsqOverlordResourceTestClient.java:
##########
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.clients.msq;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.inject.Inject;
+import org.apache.druid.guice.annotations.Json;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.http.client.HttpClient;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.msq.guice.MSQIndexingModule;
+import org.apache.druid.msq.indexing.report.MSQTaskReport;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.OverlordResourceTestClient;
+import org.apache.druid.testing.guice.TestClient;
+import org.jboss.netty.handler.codec.http.HttpMethod;
+
+import java.util.Map;
+
+/**
+ * Overlord resource client for MSQ Tasks
+ */
+public class MsqOverlordResourceTestClient extends OverlordResourceTestClient
+{
+ ObjectMapper jsonMapper;
+
+ @Inject
+ MsqOverlordResourceTestClient(
+ @Json ObjectMapper jsonMapper,
+ @TestClient HttpClient httpClient,
+ IntegrationTestingConfig config
+ )
+ {
+ super(jsonMapper, httpClient, config);
+ this.jsonMapper = jsonMapper;
+ this.jsonMapper.registerModules(new MSQIndexingModule().getJacksonModules());
+ }
+
+ public Map<String, MSQTaskReport> getTaskReportForMsqTask(String taskId)
Review Comment:
Nit: Rename to `getMsqTaskReport()` for simplicity.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] kfaraz commented on a diff in pull request #12992: Add IT for MSQ task engine using the new IT framework
Posted by GitBox <gi...@apache.org>.
kfaraz commented on code in PR #12992:
URL: https://github.com/apache/druid/pull/12992#discussion_r973102527
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/QueryResultVerifier.java:
##########
@@ -19,13 +19,19 @@
package org.apache.druid.testing.utils;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.logger.Logger;
+
import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
public class QueryResultVerifier
{
- public static boolean compareResults(
+ private static final Logger LOG = new Logger(QueryResultVerifier.class);
+
+ public static Optional<String> compareResults(
Review Comment:
Yes, within the response object, you may choose to maintain a single String that represents the message. So while `getMessage()` would simply return that message, `isSuccess()` could do a null check on it.
I would still prefer having a separate `success` flag as you might later want to include a message even in cases of success.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] paul-rogers commented on a diff in pull request #12992: Add IT for MSQ task engine using the new IT framework
Posted by GitBox <gi...@apache.org>.
paul-rogers commented on code in PR #12992:
URL: https://github.com/apache/druid/pull/12992#discussion_r973248371
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/MsqTestQueryHelper.java:
##########
@@ -0,0 +1,246 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.utils;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import org.apache.druid.indexer.TaskState;
+import org.apache.druid.indexer.TaskStatusPlus;
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.RetryUtils;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.guava.Yielder;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.msq.indexing.report.MSQResultsReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReportPayload;
+import org.apache.druid.msq.sql.SqlTaskStatus;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.sql.http.SqlQuery;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.SqlResourceTestClient;
+import org.apache.druid.testing.clients.msq.MsqOverlordResourceTestClient;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+/**
+ * Helper class to aid out ITs for MSQ.
+ * This takes all the clients required to make the necessary calls to the client APIs for MSQ and performs the boilerplate
+ * functions for the tests
+ */
+public class MsqTestQueryHelper extends AbstractTestQueryHelper<MsqQueryWithResults>
+{
+
+ private final ObjectMapper jsonMapper;
+ private final IntegrationTestingConfig config;
+ private final MsqOverlordResourceTestClient overlordClient;
+ private final SqlResourceTestClient msqClient;
+
+
+ @Inject
+ MsqTestQueryHelper(
+ final ObjectMapper jsonMapper,
+ final SqlResourceTestClient queryClient,
+ final IntegrationTestingConfig config,
+ final MsqOverlordResourceTestClient overlordClient,
+ final SqlResourceTestClient msqClient
+ )
+ {
+ super(jsonMapper, queryClient, config);
+ this.jsonMapper = jsonMapper;
+ this.config = config;
+ this.overlordClient = overlordClient;
+ this.msqClient = msqClient;
+ }
+
+ @Override
+ public String getQueryURL(String schemeAndHost)
+ {
+ return StringUtils.format("%s/druid/v2/sql/task", schemeAndHost);
+ }
+
+ /**
+ * Submits a task to the MSQ API with the given query string, and default headers and parameters
+ */
+ public String submitMsqTask(String sqlQueryString) throws ExecutionException, InterruptedException
+ {
+ return submitMsqTask(new SqlQuery(sqlQueryString, null, false, false, false, ImmutableMap.of(), null));
+ }
+
+ // Run the task, wait for it to complete, fetch the reports, verify the results,
+
+ /**
+ * Submits a {@link SqlQuery} to the MSQ API for execution. This method waits for the task to be accepted by the cluster
+ * and returns the task id associated with the submitted task
+ */
+ public String submitMsqTask(SqlQuery sqlQuery) throws ExecutionException, InterruptedException
+ {
+ String queryUrl = getQueryURL(config.getBrokerUrl());
+ Future<StatusResponseHolder> responseHolderFuture = msqClient.queryAsync(queryUrl, sqlQuery);
+ // It is okay to block here for the result because MSQ tasks return the task Id associated with it, which shouldn't
+ // consume a lot of time
+ StatusResponseHolder statusResponseHolder = responseHolderFuture.get();
Review Comment:
More to the point: this is a test. There is nothing for the test to do while the query runs. So, this code should block so the test doesn't have to include the required boilerplate. There should, however, be a timeout in case bad things happen.
##########
integration-tests/src/main/java/org/apache/druid/testing/clients/msq/MsqOverlordResourceTestClient.java:
##########
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.clients.msq;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.inject.Inject;
+import org.apache.druid.guice.annotations.Json;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.http.client.HttpClient;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.msq.guice.MSQIndexingModule;
+import org.apache.druid.msq.indexing.report.MSQTaskReport;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.OverlordResourceTestClient;
+import org.apache.druid.testing.guice.TestClient;
+import org.jboss.netty.handler.codec.http.HttpMethod;
+
+import java.util.Map;
+
+/**
+ * Overlord resource client for MSQ Tasks
+ */
+public class MsqOverlordResourceTestClient extends OverlordResourceTestClient
+{
+ ObjectMapper jsonMapper;
+
+ @Inject
+ MsqOverlordResourceTestClient(
+ @Json ObjectMapper jsonMapper,
+ @TestClient HttpClient httpClient,
+ IntegrationTestingConfig config
+ )
+ {
+ super(jsonMapper, httpClient, config);
+ this.jsonMapper = jsonMapper;
+ this.jsonMapper.registerModules(new MSQIndexingModule().getJacksonModules());
+ }
+
+ public Map<String, MSQTaskReport> getTaskReportForMsqTask(String taskId)
+ {
+ try {
+ StatusResponseHolder response = makeRequest(
+ HttpMethod.GET,
+ StringUtils.format(
+ "%s%s",
+ getIndexerURL(),
+ StringUtils.format("task/%s/reports", StringUtils.urlEncode(taskId))
+ )
+ );
+ return jsonMapper.readValue(response.getContent(), new TypeReference<Map<String, MSQTaskReport>>()
+ {
+ });
+ }
+ catch (ISE e) {
Review Comment:
This can be `RuntimeException`: no need to wrap other runtime exceptions.
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/MsqTestQueryHelper.java:
##########
@@ -0,0 +1,246 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.utils;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import org.apache.druid.indexer.TaskState;
+import org.apache.druid.indexer.TaskStatusPlus;
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.RetryUtils;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.guava.Yielder;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.msq.indexing.report.MSQResultsReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReportPayload;
+import org.apache.druid.msq.sql.SqlTaskStatus;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.sql.http.SqlQuery;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.SqlResourceTestClient;
+import org.apache.druid.testing.clients.msq.MsqOverlordResourceTestClient;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+/**
+ * Helper class to aid out ITs for MSQ.
+ * This takes all the clients required to make the necessary calls to the client APIs for MSQ and performs the boilerplate
+ * functions for the tests
+ */
+public class MsqTestQueryHelper extends AbstractTestQueryHelper<MsqQueryWithResults>
+{
+
+ private final ObjectMapper jsonMapper;
+ private final IntegrationTestingConfig config;
+ private final MsqOverlordResourceTestClient overlordClient;
+ private final SqlResourceTestClient msqClient;
+
+
+ @Inject
+ MsqTestQueryHelper(
+ final ObjectMapper jsonMapper,
+ final SqlResourceTestClient queryClient,
+ final IntegrationTestingConfig config,
+ final MsqOverlordResourceTestClient overlordClient,
+ final SqlResourceTestClient msqClient
+ )
+ {
+ super(jsonMapper, queryClient, config);
+ this.jsonMapper = jsonMapper;
+ this.config = config;
+ this.overlordClient = overlordClient;
+ this.msqClient = msqClient;
+ }
+
+ @Override
+ public String getQueryURL(String schemeAndHost)
+ {
+ return StringUtils.format("%s/druid/v2/sql/task", schemeAndHost);
+ }
+
+ /**
+ * Submits a task to the MSQ API with the given query string, and default headers and parameters
+ */
+ public String submitMsqTask(String sqlQueryString) throws ExecutionException, InterruptedException
+ {
+ return submitMsqTask(new SqlQuery(sqlQueryString, null, false, false, false, ImmutableMap.of(), null));
+ }
+
+ // Run the task, wait for it to complete, fetch the reports, verify the results,
+
+ /**
+ * Submits a {@link SqlQuery} to the MSQ API for execution. This method waits for the task to be accepted by the cluster
+ * and returns the task id associated with the submitted task
+ */
+ public String submitMsqTask(SqlQuery sqlQuery) throws ExecutionException, InterruptedException
+ {
+ String queryUrl = getQueryURL(config.getBrokerUrl());
+ Future<StatusResponseHolder> responseHolderFuture = msqClient.queryAsync(queryUrl, sqlQuery);
+ // It is okay to block here for the result because MSQ tasks return the task Id associated with it, which shouldn't
+ // consume a lot of time
+ StatusResponseHolder statusResponseHolder = responseHolderFuture.get();
+
+ // Check if the task has been accepted successfully
+ HttpResponseStatus httpResponseStatus = statusResponseHolder.getStatus();
+ if (!httpResponseStatus.equals(HttpResponseStatus.ACCEPTED)) {
+ throw new ISE(
+ "Unable to submit the task successfully. Received response status code [%d], and response content:\n[%s]",
+ httpResponseStatus,
+ statusResponseHolder.getContent()
+ );
+ }
+ String content = statusResponseHolder.getContent();
+ SqlTaskStatus sqlTaskStatus;
+ try {
+ sqlTaskStatus = jsonMapper.readValue(content, SqlTaskStatus.class);
+ }
+ catch (JsonProcessingException e) {
+ throw new ISE("Unable to parse the response");
+ }
+ if (sqlTaskStatus.getState().isFailure()) {
+ throw new ISE("Unable to start the task successfully.\nPossible exception: %s", sqlTaskStatus.getError());
+ }
+ return sqlTaskStatus.getTaskId();
+ }
+
+ /**
+ * The method retries till the task with taskId gets completed i.e. {@link TaskState#isComplete()}} returns true and
+ * returns the last fetched state {@link TaskState} of the task
+ */
+ public TaskState pollTaskIdForCompletion(String taskId) throws Exception
+ {
+ return RetryUtils.retry(
Review Comment:
Would recommend against using this utility: it masks true errors. If the server is down, it will retry 240 times.
Instead, do your own polling. A message timeout is an error here: we don't expect the server to be down. Any response other than the ones we expect is an error and need not be retried. Only the "not yet ready" response should be retried.
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/MsqTestQueryHelper.java:
##########
@@ -0,0 +1,246 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.utils;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import org.apache.druid.indexer.TaskState;
+import org.apache.druid.indexer.TaskStatusPlus;
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.RetryUtils;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.guava.Yielder;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.msq.indexing.report.MSQResultsReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReportPayload;
+import org.apache.druid.msq.sql.SqlTaskStatus;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.sql.http.SqlQuery;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.SqlResourceTestClient;
+import org.apache.druid.testing.clients.msq.MsqOverlordResourceTestClient;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+/**
+ * Helper class to aid out ITs for MSQ.
+ * This takes all the clients required to make the necessary calls to the client APIs for MSQ and performs the boilerplate
+ * functions for the tests
+ */
+public class MsqTestQueryHelper extends AbstractTestQueryHelper<MsqQueryWithResults>
+{
+
+ private final ObjectMapper jsonMapper;
+ private final IntegrationTestingConfig config;
+ private final MsqOverlordResourceTestClient overlordClient;
+ private final SqlResourceTestClient msqClient;
+
+
+ @Inject
+ MsqTestQueryHelper(
+ final ObjectMapper jsonMapper,
+ final SqlResourceTestClient queryClient,
+ final IntegrationTestingConfig config,
+ final MsqOverlordResourceTestClient overlordClient,
+ final SqlResourceTestClient msqClient
+ )
+ {
+ super(jsonMapper, queryClient, config);
+ this.jsonMapper = jsonMapper;
+ this.config = config;
+ this.overlordClient = overlordClient;
+ this.msqClient = msqClient;
+ }
+
+ @Override
+ public String getQueryURL(String schemeAndHost)
+ {
+ return StringUtils.format("%s/druid/v2/sql/task", schemeAndHost);
+ }
+
+ /**
+ * Submits a task to the MSQ API with the given query string, and default headers and parameters
+ */
+ public String submitMsqTask(String sqlQueryString) throws ExecutionException, InterruptedException
+ {
+ return submitMsqTask(new SqlQuery(sqlQueryString, null, false, false, false, ImmutableMap.of(), null));
+ }
+
+ // Run the task, wait for it to complete, fetch the reports, verify the results,
+
+ /**
+ * Submits a {@link SqlQuery} to the MSQ API for execution. This method waits for the task to be accepted by the cluster
+ * and returns the task id associated with the submitted task
+ */
+ public String submitMsqTask(SqlQuery sqlQuery) throws ExecutionException, InterruptedException
+ {
+ String queryUrl = getQueryURL(config.getBrokerUrl());
+ Future<StatusResponseHolder> responseHolderFuture = msqClient.queryAsync(queryUrl, sqlQuery);
+ // It is okay to block here for the result because MSQ tasks return the task Id associated with it, which shouldn't
+ // consume a lot of time
+ StatusResponseHolder statusResponseHolder = responseHolderFuture.get();
+
+ // Check if the task has been accepted successfully
+ HttpResponseStatus httpResponseStatus = statusResponseHolder.getStatus();
+ if (!httpResponseStatus.equals(HttpResponseStatus.ACCEPTED)) {
+ throw new ISE(
+ "Unable to submit the task successfully. Received response status code [%d], and response content:\n[%s]",
+ httpResponseStatus,
+ statusResponseHolder.getContent()
+ );
+ }
+ String content = statusResponseHolder.getContent();
+ SqlTaskStatus sqlTaskStatus;
+ try {
+ sqlTaskStatus = jsonMapper.readValue(content, SqlTaskStatus.class);
+ }
+ catch (JsonProcessingException e) {
+ throw new ISE("Unable to parse the response");
+ }
+ if (sqlTaskStatus.getState().isFailure()) {
+ throw new ISE("Unable to start the task successfully.\nPossible exception: %s", sqlTaskStatus.getError());
+ }
+ return sqlTaskStatus.getTaskId();
+ }
+
+ /**
+ * The method retries till the task with taskId gets completed i.e. {@link TaskState#isComplete()}} returns true and
+ * returns the last fetched state {@link TaskState} of the task
+ */
+ public TaskState pollTaskIdForCompletion(String taskId) throws Exception
+ {
+ return RetryUtils.retry(
+ () -> {
+ TaskStatusPlus taskStatusPlus = overlordClient.getTaskStatus(taskId);
+ TaskState statusCode = taskStatusPlus.getStatusCode();
+ if (statusCode != null && statusCode.isComplete()) {
+ return taskStatusPlus.getStatusCode();
+ }
+ throw new TaskStillRunningException();
+ },
+ (Throwable t) -> t instanceof TaskStillRunningException,
+ 100
+ );
+ }
+
+ /**
+ * Fetches status reports for a given task
+ */
+ public Map<String, MSQTaskReport> fetchStatusReports(String taskId)
+ {
+ return overlordClient.getTaskReportForMsqTask(taskId);
+ }
+
+ /**
+ * Compares the results for a given taskId. It is required that the task has produced some results that can be verified
+ */
+ private void compareResults(String taskId, MsqQueryWithResults expectedQueryWithResults)
+ {
+ Map<String, MSQTaskReport> statusReport = fetchStatusReports(taskId);
+ MSQTaskReport taskReport = statusReport.get(MSQTaskReport.REPORT_KEY);
+ if (taskReport == null) {
+ throw new ISE("Unable to fetch the status report for the task [%]", taskId);
+ }
+ MSQTaskReportPayload taskReportPayload = Preconditions.checkNotNull(
+ taskReport.getPayload(),
+ "payload"
+ );
+ MSQResultsReport resultsReport = Preconditions.checkNotNull(
+ taskReportPayload.getResults(),
+ "Results report for the task id is empty"
+ );
+
+ List<Map<String, Object>> actualResults = new ArrayList<>();
+
+ Yielder<Object[]> yielder = resultsReport.getResultYielder();
+ RowSignature rowSignature = resultsReport.getSignature();
+
+ while (!yielder.isDone()) {
+ Object[] row = yielder.get();
+ Map<String, Object> rowWithFieldNames = new LinkedHashMap<>();
+ for (int i = 0; i < row.length; ++i) {
+ rowWithFieldNames.put(rowSignature.getColumnName(i), row[i]);
+ }
+ actualResults.add(rowWithFieldNames);
+ yielder = yielder.next(null);
+ }
+
+ Optional<String> resultsComparison = QueryResultVerifier.compareResults(
+ actualResults,
+ expectedQueryWithResults.getExpectedResults(),
+ Collections.emptyList()
+ );
+ if (resultsComparison.isPresent()) {
+ throw new IAE(
+ "Expected query result is different from the actual result.\n"
+ + "Query: %s\n"
+ + "Actual Result: %s\n"
+ + "Expected Result: %s\n"
+ + "Mismatch Error: %s\n",
+ expectedQueryWithResults.getQuery(),
+ actualResults,
+ expectedQueryWithResults.getExpectedResults(),
+ resultsComparison.get()
+ );
+ }
+ }
+
+ /**
+ * Runs queries from files using MSQ and compares the results with the ones provided
+ */
+ @Override
+ public void testQueriesFromFile(String filePath, String fullDatasourcePath) throws Exception
+ {
+ LOG.info("Starting query tests for [%s]", filePath);
Review Comment:
I believe we'll find that the file format here will be quite hard to maintain. Looks like it is JSON, which contains SQL, which contains JSON. While JSON is handy, this is a place where it does not serve us well.
There is an open PR that proposes an "Impala-like" file format for query tests:
```text
=== sql
SELECT ...
=== results
{"a": 10, "b": "foo"}
...
```
It is too much to ask to include that format here: the test PR is not yet merged. However, moving forward, perhaps we can retrofit that format here to make the query files a bit easier to use.
In fact, we could use the entire framework: it already has mechanisms to compare expected and actual results, to loop through test cases, to run queries with differing context options, etc.
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/QueryResultVerifier.java:
##########
@@ -19,13 +19,19 @@
package org.apache.druid.testing.utils;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.logger.Logger;
+
import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
public class QueryResultVerifier
{
- public static boolean compareResults(
+ private static final Logger LOG = new Logger(QueryResultVerifier.class);
+
+ public static Optional<String> compareResults(
Review Comment:
The approach taken in the planner test framework is to define a "test case" that has the query and expected results. Run that to get the actual results. Compare the two. The idea is to focus on the test and make the test code as simple as possible. Here:
```java
testCase = loadTestCase(...);
result = testCase.run(...);
```
Where `run()` would verify the results and would fail if the results don't match. (Results could be an expected error, or expected rows.)
##########
integration-tests/src/main/java/org/apache/druid/testing/clients/msq/MsqOverlordResourceTestClient.java:
##########
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.clients.msq;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.inject.Inject;
+import org.apache.druid.guice.annotations.Json;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.http.client.HttpClient;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.msq.guice.MSQIndexingModule;
+import org.apache.druid.msq.indexing.report.MSQTaskReport;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.OverlordResourceTestClient;
+import org.apache.druid.testing.guice.TestClient;
+import org.jboss.netty.handler.codec.http.HttpMethod;
+
+import java.util.Map;
+
+/**
+ * Overlord resource client for MSQ Tasks
+ */
+public class MsqOverlordResourceTestClient extends OverlordResourceTestClient
+{
+ ObjectMapper jsonMapper;
+
+ @Inject
+ MsqOverlordResourceTestClient(
+ @Json ObjectMapper jsonMapper,
+ @TestClient HttpClient httpClient,
+ IntegrationTestingConfig config
+ )
+ {
+ super(jsonMapper, httpClient, config);
+ this.jsonMapper = jsonMapper;
+ this.jsonMapper.registerModules(new MSQIndexingModule().getJacksonModules());
+ }
+
+ public Map<String, MSQTaskReport> getTaskReportForMsqTask(String taskId)
+ {
+ try {
+ StatusResponseHolder response = makeRequest(
+ HttpMethod.GET,
+ StringUtils.format(
+ "%s%s",
+ getIndexerURL(),
+ StringUtils.format("task/%s/reports", StringUtils.urlEncode(taskId))
+ )
+ );
Review Comment:
It is a real nuisance that, with the dozens of places that the "format JSON message, send it, and convert the response" code is used, we don't have a reusable class. In one of the branches I started to create such a thing; let me see if I can find it and perhaps we can use it here to avoid redundancy.
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/MsqTestQueryHelper.java:
##########
@@ -0,0 +1,246 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.utils;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import org.apache.druid.indexer.TaskState;
+import org.apache.druid.indexer.TaskStatusPlus;
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.RetryUtils;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.guava.Yielder;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.msq.indexing.report.MSQResultsReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReportPayload;
+import org.apache.druid.msq.sql.SqlTaskStatus;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.sql.http.SqlQuery;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.SqlResourceTestClient;
+import org.apache.druid.testing.clients.msq.MsqOverlordResourceTestClient;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+/**
+ * Helper class to aid out ITs for MSQ.
+ * This takes all the clients required to make the necessary calls to the client APIs for MSQ and performs the boilerplate
+ * functions for the tests
+ */
+public class MsqTestQueryHelper extends AbstractTestQueryHelper<MsqQueryWithResults>
+{
+
+ private final ObjectMapper jsonMapper;
+ private final IntegrationTestingConfig config;
+ private final MsqOverlordResourceTestClient overlordClient;
+ private final SqlResourceTestClient msqClient;
+
+
+ @Inject
+ MsqTestQueryHelper(
+ final ObjectMapper jsonMapper,
+ final SqlResourceTestClient queryClient,
+ final IntegrationTestingConfig config,
+ final MsqOverlordResourceTestClient overlordClient,
+ final SqlResourceTestClient msqClient
+ )
+ {
+ super(jsonMapper, queryClient, config);
+ this.jsonMapper = jsonMapper;
+ this.config = config;
+ this.overlordClient = overlordClient;
+ this.msqClient = msqClient;
+ }
+
+ @Override
+ public String getQueryURL(String schemeAndHost)
+ {
+ return StringUtils.format("%s/druid/v2/sql/task", schemeAndHost);
+ }
+
+ /**
+ * Submits a task to the MSQ API with the given query string, and default headers and parameters
+ */
+ public String submitMsqTask(String sqlQueryString) throws ExecutionException, InterruptedException
+ {
+ return submitMsqTask(new SqlQuery(sqlQueryString, null, false, false, false, ImmutableMap.of(), null));
+ }
+
+ // Run the task, wait for it to complete, fetch the reports, verify the results,
+
+ /**
+ * Submits a {@link SqlQuery} to the MSQ API for execution. This method waits for the task to be accepted by the cluster
+ * and returns the task id associated with the submitted task
+ */
+ public String submitMsqTask(SqlQuery sqlQuery) throws ExecutionException, InterruptedException
+ {
+ String queryUrl = getQueryURL(config.getBrokerUrl());
+ Future<StatusResponseHolder> responseHolderFuture = msqClient.queryAsync(queryUrl, sqlQuery);
+ // It is okay to block here for the result because MSQ tasks return the task Id associated with it, which shouldn't
+ // consume a lot of time
+ StatusResponseHolder statusResponseHolder = responseHolderFuture.get();
+
+ // Check if the task has been accepted successfully
+ HttpResponseStatus httpResponseStatus = statusResponseHolder.getStatus();
+ if (!httpResponseStatus.equals(HttpResponseStatus.ACCEPTED)) {
+ throw new ISE(
+ "Unable to submit the task successfully. Received response status code [%d], and response content:\n[%s]",
+ httpResponseStatus,
+ statusResponseHolder.getContent()
+ );
+ }
+ String content = statusResponseHolder.getContent();
+ SqlTaskStatus sqlTaskStatus;
+ try {
+ sqlTaskStatus = jsonMapper.readValue(content, SqlTaskStatus.class);
+ }
+ catch (JsonProcessingException e) {
+ throw new ISE("Unable to parse the response");
+ }
+ if (sqlTaskStatus.getState().isFailure()) {
Review Comment:
Perhaps return the response, error or not. This lets us write tests that verify that something fails when it should. For example, if we want to verify permissions or that the engine rejects certain invalid syntax, etc.
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/AbstractTestQueryHelper.java:
##########
@@ -146,24 +147,29 @@ private void testQueries(String url, List<QueryResultType> queries) throws Excep
for (QueryResultType queryWithResult : queries) {
LOG.info("Running Query %s", queryWithResult.getQuery());
List<Map<String, Object>> result = queryClient.query(url, queryWithResult.getQuery());
- if (!QueryResultVerifier.compareResults(result, queryWithResult.getExpectedResults(),
- queryWithResult.getFieldsToTest()
- )) {
+ Optional<String> resultsComparison = QueryResultVerifier.compareResults(result, queryWithResult.getExpectedResults(),
+ queryWithResult.getFieldsToTest());
+ if (resultsComparison.isPresent()) {
LOG.error(
"Failed while executing query %s \n expectedResults: %s \n actualResults : %s",
queryWithResult.getQuery(),
jsonMapper.writeValueAsString(queryWithResult.getExpectedResults()),
jsonMapper.writeValueAsString(result)
);
- failed = true;
+ Assert.fail(StringUtils.format(
+ "Results mismatch while executing the query %s.\n"
+ + "Expected results: %s\n"
+ + "Actual results: %s\n"
+ + "Mismatch error: %s\n",
+ queryWithResult.getQuery(),
+ jsonMapper.writeValueAsString(queryWithResult.getExpectedResults()),
+ jsonMapper.writeValueAsString(result),
+ resultsComparison.get()
+ ));
Review Comment:
The above will be very hard to debug: the JSON payloads are large. This is already a problem in SQL tests when we compare queries: if they don't match, we're left with two big JSON blobs and it is tedious to determine where they differ.
Results are a set of rows. So, it would be great to report, say, the number of the first row that differs, and display just those values. Then, display the actual results in the same format that they will appear in the "expected results" file. This will allow us to develop a new test by starting with empty results, then copy the failed test results into the file, then watch the test pass.
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/MsqTestQueryHelper.java:
##########
@@ -0,0 +1,246 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.utils;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import org.apache.druid.indexer.TaskState;
+import org.apache.druid.indexer.TaskStatusPlus;
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.RetryUtils;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.guava.Yielder;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.msq.indexing.report.MSQResultsReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReportPayload;
+import org.apache.druid.msq.sql.SqlTaskStatus;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.sql.http.SqlQuery;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.SqlResourceTestClient;
+import org.apache.druid.testing.clients.msq.MsqOverlordResourceTestClient;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+/**
+ * Helper class to aid out ITs for MSQ.
+ * This takes all the clients required to make the necessary calls to the client APIs for MSQ and performs the boilerplate
+ * functions for the tests
+ */
+public class MsqTestQueryHelper extends AbstractTestQueryHelper<MsqQueryWithResults>
+{
+
+ private final ObjectMapper jsonMapper;
+ private final IntegrationTestingConfig config;
+ private final MsqOverlordResourceTestClient overlordClient;
+ private final SqlResourceTestClient msqClient;
+
+
+ @Inject
+ MsqTestQueryHelper(
+ final ObjectMapper jsonMapper,
+ final SqlResourceTestClient queryClient,
+ final IntegrationTestingConfig config,
+ final MsqOverlordResourceTestClient overlordClient,
+ final SqlResourceTestClient msqClient
+ )
+ {
+ super(jsonMapper, queryClient, config);
+ this.jsonMapper = jsonMapper;
+ this.config = config;
+ this.overlordClient = overlordClient;
+ this.msqClient = msqClient;
+ }
+
+ @Override
+ public String getQueryURL(String schemeAndHost)
+ {
+ return StringUtils.format("%s/druid/v2/sql/task", schemeAndHost);
+ }
+
+ /**
+ * Submits a task to the MSQ API with the given query string, and default headers and parameters
+ */
+ public String submitMsqTask(String sqlQueryString) throws ExecutionException, InterruptedException
+ {
+ return submitMsqTask(new SqlQuery(sqlQueryString, null, false, false, false, ImmutableMap.of(), null));
+ }
+
+ // Run the task, wait for it to complete, fetch the reports, verify the results,
+
+ /**
+ * Submits a {@link SqlQuery} to the MSQ API for execution. This method waits for the task to be accepted by the cluster
+ * and returns the task id associated with the submitted task
+ */
+ public String submitMsqTask(SqlQuery sqlQuery) throws ExecutionException, InterruptedException
+ {
+ String queryUrl = getQueryURL(config.getBrokerUrl());
+ Future<StatusResponseHolder> responseHolderFuture = msqClient.queryAsync(queryUrl, sqlQuery);
+ // It is okay to block here for the result because MSQ tasks return the task Id associated with it, which shouldn't
+ // consume a lot of time
+ StatusResponseHolder statusResponseHolder = responseHolderFuture.get();
+
+ // Check if the task has been accepted successfully
+ HttpResponseStatus httpResponseStatus = statusResponseHolder.getStatus();
+ if (!httpResponseStatus.equals(HttpResponseStatus.ACCEPTED)) {
+ throw new ISE(
+ "Unable to submit the task successfully. Received response status code [%d], and response content:\n[%s]",
+ httpResponseStatus,
+ statusResponseHolder.getContent()
+ );
+ }
+ String content = statusResponseHolder.getContent();
+ SqlTaskStatus sqlTaskStatus;
+ try {
+ sqlTaskStatus = jsonMapper.readValue(content, SqlTaskStatus.class);
+ }
+ catch (JsonProcessingException e) {
+ throw new ISE("Unable to parse the response");
+ }
+ if (sqlTaskStatus.getState().isFailure()) {
+ throw new ISE("Unable to start the task successfully.\nPossible exception: %s", sqlTaskStatus.getError());
+ }
+ return sqlTaskStatus.getTaskId();
+ }
+
+ /**
+ * The method retries till the task with taskId gets completed i.e. {@link TaskState#isComplete()}} returns true and
+ * returns the last fetched state {@link TaskState} of the task
+ */
+ public TaskState pollTaskIdForCompletion(String taskId) throws Exception
+ {
+ return RetryUtils.retry(
+ () -> {
+ TaskStatusPlus taskStatusPlus = overlordClient.getTaskStatus(taskId);
+ TaskState statusCode = taskStatusPlus.getStatusCode();
+ if (statusCode != null && statusCode.isComplete()) {
+ return taskStatusPlus.getStatusCode();
+ }
+ throw new TaskStillRunningException();
+ },
+ (Throwable t) -> t instanceof TaskStillRunningException,
+ 100
+ );
+ }
+
+ /**
+ * Fetches status reports for a given task
+ */
+ public Map<String, MSQTaskReport> fetchStatusReports(String taskId)
+ {
+ return overlordClient.getTaskReportForMsqTask(taskId);
+ }
+
+ /**
+ * Compares the results for a given taskId. It is required that the task has produced some results that can be verified
+ */
+ private void compareResults(String taskId, MsqQueryWithResults expectedQueryWithResults)
+ {
+ Map<String, MSQTaskReport> statusReport = fetchStatusReports(taskId);
+ MSQTaskReport taskReport = statusReport.get(MSQTaskReport.REPORT_KEY);
+ if (taskReport == null) {
+ throw new ISE("Unable to fetch the status report for the task [%]", taskId);
+ }
+ MSQTaskReportPayload taskReportPayload = Preconditions.checkNotNull(
+ taskReport.getPayload(),
+ "payload"
+ );
+ MSQResultsReport resultsReport = Preconditions.checkNotNull(
+ taskReportPayload.getResults(),
+ "Results report for the task id is empty"
+ );
+
+ List<Map<String, Object>> actualResults = new ArrayList<>();
+
+ Yielder<Object[]> yielder = resultsReport.getResultYielder();
+ RowSignature rowSignature = resultsReport.getSignature();
+
+ while (!yielder.isDone()) {
+ Object[] row = yielder.get();
+ Map<String, Object> rowWithFieldNames = new LinkedHashMap<>();
+ for (int i = 0; i < row.length; ++i) {
+ rowWithFieldNames.put(rowSignature.getColumnName(i), row[i]);
+ }
+ actualResults.add(rowWithFieldNames);
+ yielder = yielder.next(null);
+ }
+
+ Optional<String> resultsComparison = QueryResultVerifier.compareResults(
+ actualResults,
+ expectedQueryWithResults.getExpectedResults(),
+ Collections.emptyList()
+ );
+ if (resultsComparison.isPresent()) {
+ throw new IAE(
+ "Expected query result is different from the actual result.\n"
+ + "Query: %s\n"
+ + "Actual Result: %s\n"
+ + "Expected Result: %s\n"
+ + "Mismatch Error: %s\n",
Review Comment:
See not above about format.
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/MsqTestQueryHelper.java:
##########
@@ -0,0 +1,246 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.utils;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import org.apache.druid.indexer.TaskState;
+import org.apache.druid.indexer.TaskStatusPlus;
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.RetryUtils;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.guava.Yielder;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.msq.indexing.report.MSQResultsReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReportPayload;
+import org.apache.druid.msq.sql.SqlTaskStatus;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.sql.http.SqlQuery;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.SqlResourceTestClient;
+import org.apache.druid.testing.clients.msq.MsqOverlordResourceTestClient;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+/**
+ * Helper class to aid out ITs for MSQ.
+ * This takes all the clients required to make the necessary calls to the client APIs for MSQ and performs the boilerplate
+ * functions for the tests
+ */
+public class MsqTestQueryHelper extends AbstractTestQueryHelper<MsqQueryWithResults>
+{
+
+ private final ObjectMapper jsonMapper;
+ private final IntegrationTestingConfig config;
+ private final MsqOverlordResourceTestClient overlordClient;
+ private final SqlResourceTestClient msqClient;
+
+
+ @Inject
+ MsqTestQueryHelper(
+ final ObjectMapper jsonMapper,
+ final SqlResourceTestClient queryClient,
+ final IntegrationTestingConfig config,
+ final MsqOverlordResourceTestClient overlordClient,
+ final SqlResourceTestClient msqClient
+ )
+ {
+ super(jsonMapper, queryClient, config);
+ this.jsonMapper = jsonMapper;
+ this.config = config;
+ this.overlordClient = overlordClient;
+ this.msqClient = msqClient;
+ }
+
+ @Override
+ public String getQueryURL(String schemeAndHost)
+ {
+ return StringUtils.format("%s/druid/v2/sql/task", schemeAndHost);
+ }
+
+ /**
+ * Submits a task to the MSQ API with the given query string, and default headers and parameters
+ */
+ public String submitMsqTask(String sqlQueryString) throws ExecutionException, InterruptedException
+ {
+ return submitMsqTask(new SqlQuery(sqlQueryString, null, false, false, false, ImmutableMap.of(), null));
+ }
+
+ // Run the task, wait for it to complete, fetch the reports, verify the results,
+
+ /**
+ * Submits a {@link SqlQuery} to the MSQ API for execution. This method waits for the task to be accepted by the cluster
+ * and returns the task id associated with the submitted task
+ */
+ public String submitMsqTask(SqlQuery sqlQuery) throws ExecutionException, InterruptedException
+ {
+ String queryUrl = getQueryURL(config.getBrokerUrl());
+ Future<StatusResponseHolder> responseHolderFuture = msqClient.queryAsync(queryUrl, sqlQuery);
+ // It is okay to block here for the result because MSQ tasks return the task Id associated with it, which shouldn't
+ // consume a lot of time
+ StatusResponseHolder statusResponseHolder = responseHolderFuture.get();
+
+ // Check if the task has been accepted successfully
+ HttpResponseStatus httpResponseStatus = statusResponseHolder.getStatus();
+ if (!httpResponseStatus.equals(HttpResponseStatus.ACCEPTED)) {
+ throw new ISE(
+ "Unable to submit the task successfully. Received response status code [%d], and response content:\n[%s]",
+ httpResponseStatus,
+ statusResponseHolder.getContent()
+ );
+ }
+ String content = statusResponseHolder.getContent();
+ SqlTaskStatus sqlTaskStatus;
+ try {
+ sqlTaskStatus = jsonMapper.readValue(content, SqlTaskStatus.class);
+ }
+ catch (JsonProcessingException e) {
+ throw new ISE("Unable to parse the response");
+ }
+ if (sqlTaskStatus.getState().isFailure()) {
+ throw new ISE("Unable to start the task successfully.\nPossible exception: %s", sqlTaskStatus.getError());
+ }
+ return sqlTaskStatus.getTaskId();
+ }
+
+ /**
+ * The method retries till the task with taskId gets completed i.e. {@link TaskState#isComplete()}} returns true and
+ * returns the last fetched state {@link TaskState} of the task
+ */
+ public TaskState pollTaskIdForCompletion(String taskId) throws Exception
+ {
+ return RetryUtils.retry(
+ () -> {
+ TaskStatusPlus taskStatusPlus = overlordClient.getTaskStatus(taskId);
+ TaskState statusCode = taskStatusPlus.getStatusCode();
+ if (statusCode != null && statusCode.isComplete()) {
+ return taskStatusPlus.getStatusCode();
+ }
+ throw new TaskStillRunningException();
+ },
+ (Throwable t) -> t instanceof TaskStillRunningException,
+ 100
+ );
+ }
+
+ /**
+ * Fetches status reports for a given task
+ */
+ public Map<String, MSQTaskReport> fetchStatusReports(String taskId)
Review Comment:
There is only ever one report for MSQ, correct? I've never seen more than one.
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/MsqTestQueryHelper.java:
##########
@@ -0,0 +1,246 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.utils;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import org.apache.druid.indexer.TaskState;
+import org.apache.druid.indexer.TaskStatusPlus;
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.RetryUtils;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.guava.Yielder;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.msq.indexing.report.MSQResultsReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReportPayload;
+import org.apache.druid.msq.sql.SqlTaskStatus;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.sql.http.SqlQuery;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.SqlResourceTestClient;
+import org.apache.druid.testing.clients.msq.MsqOverlordResourceTestClient;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+/**
+ * Helper class to aid out ITs for MSQ.
+ * This takes all the clients required to make the necessary calls to the client APIs for MSQ and performs the boilerplate
+ * functions for the tests
+ */
+public class MsqTestQueryHelper extends AbstractTestQueryHelper<MsqQueryWithResults>
+{
+
+ private final ObjectMapper jsonMapper;
+ private final IntegrationTestingConfig config;
+ private final MsqOverlordResourceTestClient overlordClient;
+ private final SqlResourceTestClient msqClient;
+
+
+ @Inject
+ MsqTestQueryHelper(
+ final ObjectMapper jsonMapper,
+ final SqlResourceTestClient queryClient,
+ final IntegrationTestingConfig config,
+ final MsqOverlordResourceTestClient overlordClient,
+ final SqlResourceTestClient msqClient
+ )
+ {
+ super(jsonMapper, queryClient, config);
+ this.jsonMapper = jsonMapper;
+ this.config = config;
+ this.overlordClient = overlordClient;
+ this.msqClient = msqClient;
+ }
+
+ @Override
+ public String getQueryURL(String schemeAndHost)
+ {
+ return StringUtils.format("%s/druid/v2/sql/task", schemeAndHost);
+ }
+
+ /**
+ * Submits a task to the MSQ API with the given query string, and default headers and parameters
+ */
+ public String submitMsqTask(String sqlQueryString) throws ExecutionException, InterruptedException
+ {
+ return submitMsqTask(new SqlQuery(sqlQueryString, null, false, false, false, ImmutableMap.of(), null));
+ }
+
+ // Run the task, wait for it to complete, fetch the reports, verify the results,
+
+ /**
+ * Submits a {@link SqlQuery} to the MSQ API for execution. This method waits for the task to be accepted by the cluster
+ * and returns the task id associated with the submitted task
+ */
+ public String submitMsqTask(SqlQuery sqlQuery) throws ExecutionException, InterruptedException
+ {
+ String queryUrl = getQueryURL(config.getBrokerUrl());
+ Future<StatusResponseHolder> responseHolderFuture = msqClient.queryAsync(queryUrl, sqlQuery);
+ // It is okay to block here for the result because MSQ tasks return the task Id associated with it, which shouldn't
+ // consume a lot of time
+ StatusResponseHolder statusResponseHolder = responseHolderFuture.get();
+
+ // Check if the task has been accepted successfully
+ HttpResponseStatus httpResponseStatus = statusResponseHolder.getStatus();
+ if (!httpResponseStatus.equals(HttpResponseStatus.ACCEPTED)) {
+ throw new ISE(
+ "Unable to submit the task successfully. Received response status code [%d], and response content:\n[%s]",
+ httpResponseStatus,
+ statusResponseHolder.getContent()
+ );
+ }
+ String content = statusResponseHolder.getContent();
+ SqlTaskStatus sqlTaskStatus;
+ try {
+ sqlTaskStatus = jsonMapper.readValue(content, SqlTaskStatus.class);
+ }
+ catch (JsonProcessingException e) {
+ throw new ISE("Unable to parse the response");
+ }
+ if (sqlTaskStatus.getState().isFailure()) {
+ throw new ISE("Unable to start the task successfully.\nPossible exception: %s", sqlTaskStatus.getError());
+ }
+ return sqlTaskStatus.getTaskId();
+ }
+
+ /**
+ * The method retries till the task with taskId gets completed i.e. {@link TaskState#isComplete()}} returns true and
+ * returns the last fetched state {@link TaskState} of the task
+ */
+ public TaskState pollTaskIdForCompletion(String taskId) throws Exception
+ {
+ return RetryUtils.retry(
+ () -> {
+ TaskStatusPlus taskStatusPlus = overlordClient.getTaskStatus(taskId);
+ TaskState statusCode = taskStatusPlus.getStatusCode();
+ if (statusCode != null && statusCode.isComplete()) {
+ return taskStatusPlus.getStatusCode();
+ }
+ throw new TaskStillRunningException();
+ },
+ (Throwable t) -> t instanceof TaskStillRunningException,
+ 100
+ );
+ }
+
+ /**
+ * Fetches status reports for a given task
+ */
+ public Map<String, MSQTaskReport> fetchStatusReports(String taskId)
Review Comment:
This would be a handy place to have a `MsqResult` object that holds the reports and pulls out the relevant bits as needed.
##########
integration-tests-ex/cases/src/test/resources/multi-stage-query/wikipedia_msq_select_query1.json:
##########
@@ -0,0 +1,31 @@
+[
Review Comment:
The above file is empty. Intended?
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/QueryResultVerifier.java:
##########
@@ -19,13 +19,19 @@
package org.apache.druid.testing.utils;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.logger.Logger;
+
import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
public class QueryResultVerifier
{
- public static boolean compareResults(
+ private static final Logger LOG = new Logger(QueryResultVerifier.class);
+
+ public static Optional<String> compareResults(
Review Comment:
Since this is used in a test, we won't do anything with the results. The call will be either:
```java
Assert.assertTrue(QueryResultVerifier.compareResults(...));
```
Or just:
```java
QueryResultVerifier.compareResults(...);
```
Given the fact that this is a test, perhaps rename the method:
```java
results = // run the query
QueryResultVerifier.assertEqual(...);
```
##########
integration-tests/src/main/java/org/apache/druid/testing/clients/msq/MsqOverlordResourceTestClient.java:
##########
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.clients.msq;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.inject.Inject;
+import org.apache.druid.guice.annotations.Json;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.http.client.HttpClient;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.msq.guice.MSQIndexingModule;
+import org.apache.druid.msq.indexing.report.MSQTaskReport;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.OverlordResourceTestClient;
+import org.apache.druid.testing.guice.TestClient;
+import org.jboss.netty.handler.codec.http.HttpMethod;
+
+import java.util.Map;
+
+/**
+ * Overlord resource client for MSQ Tasks
+ */
+public class MsqOverlordResourceTestClient extends OverlordResourceTestClient
Review Comment:
I wonder if these should simply be merged into the Overlord client. Or, be add-ons that take the overlord client as a parameter. Otherwise, we may find ourselves in need of Overlord Client X and Overlord Client Y, both as subclasses, and thus need two client when one would do.
##########
integration-tests/src/main/java/org/apache/druid/testing/clients/msq/MsqOverlordResourceTestClient.java:
##########
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.clients.msq;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.inject.Inject;
+import org.apache.druid.guice.annotations.Json;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.http.client.HttpClient;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.msq.guice.MSQIndexingModule;
+import org.apache.druid.msq.indexing.report.MSQTaskReport;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.OverlordResourceTestClient;
+import org.apache.druid.testing.guice.TestClient;
+import org.jboss.netty.handler.codec.http.HttpMethod;
+
+import java.util.Map;
+
+/**
+ * Overlord resource client for MSQ Tasks
+ */
+public class MsqOverlordResourceTestClient extends OverlordResourceTestClient
+{
+ ObjectMapper jsonMapper;
+
+ @Inject
+ MsqOverlordResourceTestClient(
+ @Json ObjectMapper jsonMapper,
+ @TestClient HttpClient httpClient,
+ IntegrationTestingConfig config
+ )
+ {
+ super(jsonMapper, httpClient, config);
+ this.jsonMapper = jsonMapper;
+ this.jsonMapper.registerModules(new MSQIndexingModule().getJacksonModules());
+ }
+
+ public Map<String, MSQTaskReport> getTaskReportForMsqTask(String taskId)
Review Comment:
At present, MSQ is somewhat awkward when running a SELECT: one has to paw through the reports to get the results. One can safely predict that this will change later to have some direct way to get results. So, let's not bake the current report-based results too deeply into the tests.
To anticipate this, perhaps the code here should have a method to return just results. Or, a `MsqResponse` object that holds reports and lets us pull out the error (if any) and the results. The results would ideally include schema and data (pulled from the various places in the reports.)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] cryptoe commented on a diff in pull request #12992: Add IT for MSQ task engine using the new IT framework
Posted by GitBox <gi...@apache.org>.
cryptoe commented on code in PR #12992:
URL: https://github.com/apache/druid/pull/12992#discussion_r956916880
##########
integration-tests/src/main/java/org/apache/druid/testing/clients/MsqTestClient.java:
##########
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.clients;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.inject.Inject;
+import org.apache.druid.java.util.http.client.HttpClient;
+import org.apache.druid.sql.http.SqlQuery;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.guice.TestClient;
+
+import javax.ws.rs.core.MediaType;
+
+public class MsqTestClient extends AbstractQueryResourceTestClient<SqlQuery>
Review Comment:
I meant "SqlResourceTestClient". Having 2 concrete classes having exactly the functionality seems weird to me.
Just change the name of the variable no?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] LakshSingla commented on a diff in pull request #12992: Add IT for MSQ task engine using the new IT framework
Posted by GitBox <gi...@apache.org>.
LakshSingla commented on code in PR #12992:
URL: https://github.com/apache/druid/pull/12992#discussion_r973844721
##########
integration-tests/src/main/java/org/apache/druid/testing/clients/msq/MsqOverlordResourceTestClient.java:
##########
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.clients.msq;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.inject.Inject;
+import org.apache.druid.guice.annotations.Json;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.http.client.HttpClient;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.msq.guice.MSQIndexingModule;
+import org.apache.druid.msq.indexing.report.MSQTaskReport;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.OverlordResourceTestClient;
+import org.apache.druid.testing.guice.TestClient;
+import org.jboss.netty.handler.codec.http.HttpMethod;
+
+import java.util.Map;
+
+/**
+ * Overlord resource client for MSQ Tasks
+ */
+public class MsqOverlordResourceTestClient extends OverlordResourceTestClient
+{
+ ObjectMapper jsonMapper;
+
+ @Inject
+ MsqOverlordResourceTestClient(
+ @Json ObjectMapper jsonMapper,
+ @TestClient HttpClient httpClient,
+ IntegrationTestingConfig config
+ )
+ {
+ super(jsonMapper, httpClient, config);
+ this.jsonMapper = jsonMapper;
+ this.jsonMapper.registerModules(new MSQIndexingModule().getJacksonModules());
+ }
+
+ public Map<String, MSQTaskReport> getTaskReportForMsqTask(String taskId)
+ {
+ try {
+ StatusResponseHolder response = makeRequest(
+ HttpMethod.GET,
+ StringUtils.format(
+ "%s%s",
+ getIndexerURL(),
+ StringUtils.format("task/%s/reports", StringUtils.urlEncode(taskId))
+ )
+ );
Review Comment:
`Map<String, TaskReport>` is supposed to be an overlord thing with how it stores the task reports wherein the key is supposed to be the `REPORT_KEY`. For multiStageQuery it is going to `multiStageQuery`.
Yeah, I agree that it is a bit weird because the map generally contains a single task report. I remember wanting to use the generics to drop the use of this class, but that was causing a Jackson serde error because it was unable to figure out the subtype correctly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] kfaraz commented on a diff in pull request #12992: Add IT for MSQ task engine using the new IT framework
Posted by GitBox <gi...@apache.org>.
kfaraz commented on code in PR #12992:
URL: https://github.com/apache/druid/pull/12992#discussion_r964590635
##########
integration-tests/src/main/java/org/apache/druid/testing/clients/msq/MsqOverlordResourceTestClient.java:
##########
@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.clients.msq;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableList;
+import com.google.inject.Inject;
+import org.apache.druid.guice.annotations.Json;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.http.client.HttpClient;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.msq.guice.MSQIndexingModule;
+import org.apache.druid.msq.indexing.report.MSQTaskReport;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.OverlordResourceTestClient;
+import org.apache.druid.testing.guice.TestClient;
+import org.jboss.netty.handler.codec.http.HttpMethod;
+
+import java.util.Map;
+
+/**
+ * Overlord resource client for MSQ Tasks
+ */
+public class MsqOverlordResourceTestClient extends OverlordResourceTestClient
+{
+ ObjectMapper jsonMapper;
Review Comment:
Nit: Please make this private if we don't need to expose it.
##########
integration-tests-ex/cases/cluster/Common/environment-configs/common.env:
##########
@@ -49,7 +49,7 @@ DRUID_INSTANCE=
# variables: druid_standard_loadList defined here, and druid_test_loadList, defined
# in a docker-compose.yaml file, for any test-specific extensions.
# See compose.md for more details.
-druid_standard_loadList=mysql-metadata-storage,it-tools,druid-lookups-cached-global,druid-histogram,druid-datasketches,druid-parquet-extensions,druid-avro-extensions,druid-protobuf-extensions,druid-orc-extensions,druid-kafka-indexing-service,druid-s3-extensions
+druid_standard_loadList=mysql-metadata-storage,it-tools,druid-lookups-cached-global,druid-histogram,druid-datasketches,druid-parquet-extensions,druid-avro-extensions,druid-protobuf-extensions,druid-orc-extensions,druid-kafka-indexing-service,druid-s3-extensions,druid-multi-stage-query
Review Comment:
Does this mean that the `multi-stage-query` extension will be loaded for every integration test setup?
Is there a way to load it just for the multi stage query tests?
##########
integration-tests/src/main/java/org/apache/druid/testing/clients/msq/MsqOverlordResourceTestClient.java:
##########
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.clients.msq;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.inject.Inject;
+import org.apache.druid.guice.annotations.Json;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.http.client.HttpClient;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.msq.guice.MSQIndexingModule;
+import org.apache.druid.msq.indexing.report.MSQTaskReport;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.OverlordResourceTestClient;
+import org.apache.druid.testing.guice.TestClient;
+import org.jboss.netty.handler.codec.http.HttpMethod;
+
+import java.util.Map;
+
+/**
+ * Overlord resource client for MSQ Tasks
+ */
+public class MsqOverlordResourceTestClient extends OverlordResourceTestClient
Review Comment:
These MSQ specific test utilities such as `MsqOverlordResourceTestClient`, `MsqQueryWithResults`, `MsqTestQueryHelper` will be used only for MSQ ITs only. So they should live in the same place as the `ITMultiStageQuery` itself.
##########
integration-tests/src/main/java/org/apache/druid/testing/clients/msq/MsqOverlordResourceTestClient.java:
##########
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.clients.msq;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.inject.Inject;
+import org.apache.druid.guice.annotations.Json;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.http.client.HttpClient;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.msq.guice.MSQIndexingModule;
+import org.apache.druid.msq.indexing.report.MSQTaskReport;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.OverlordResourceTestClient;
+import org.apache.druid.testing.guice.TestClient;
+import org.jboss.netty.handler.codec.http.HttpMethod;
+
+import java.util.Map;
+
+/**
+ * Overlord resource client for MSQ Tasks
+ */
+public class MsqOverlordResourceTestClient extends OverlordResourceTestClient
+{
+ ObjectMapper jsonMapper;
+
+ @Inject
+ MsqOverlordResourceTestClient(
+ @Json ObjectMapper jsonMapper,
+ @TestClient HttpClient httpClient,
+ IntegrationTestingConfig config
+ )
+ {
+ super(jsonMapper, httpClient, config);
+ this.jsonMapper = jsonMapper;
+ this.jsonMapper.registerModules(new MSQIndexingModule().getJacksonModules());
+ }
+
+ public Map<String, MSQTaskReport> getTaskReportForMsqTask(String taskId)
Review Comment:
Nit: Rename to `getMsqTaskReport()`.
##########
integration-tests/src/main/java/org/apache/druid/testing/clients/msq/MsqOverlordResourceTestClient.java:
##########
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.clients.msq;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.inject.Inject;
+import org.apache.druid.guice.annotations.Json;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.http.client.HttpClient;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.msq.guice.MSQIndexingModule;
+import org.apache.druid.msq.indexing.report.MSQTaskReport;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.OverlordResourceTestClient;
+import org.apache.druid.testing.guice.TestClient;
+import org.jboss.netty.handler.codec.http.HttpMethod;
+
+import java.util.Map;
+
+/**
+ * Overlord resource client for MSQ Tasks
+ */
+public class MsqOverlordResourceTestClient extends OverlordResourceTestClient
+{
+ ObjectMapper jsonMapper;
+
+ @Inject
+ MsqOverlordResourceTestClient(
+ @Json ObjectMapper jsonMapper,
+ @TestClient HttpClient httpClient,
+ IntegrationTestingConfig config
+ )
+ {
+ super(jsonMapper, httpClient, config);
+ this.jsonMapper = jsonMapper;
+ this.jsonMapper.registerModules(new MSQIndexingModule().getJacksonModules());
+ }
+
+ public Map<String, MSQTaskReport> getTaskReportForMsqTask(String taskId)
+ {
+ try {
+ StatusResponseHolder response = makeRequest(
+ HttpMethod.GET,
+ StringUtils.format(
+ "%s%s",
+ getIndexerURL(),
+ StringUtils.format("task/%s/reports", StringUtils.urlEncode(taskId))
+ )
+ );
Review Comment:
Nit: This part seems to be duplicated from `OverlordResourceTestClient`. Maybe have a common method for this.
What is supposed to be the key of this map? We should add some javadoc to clarify that.
I would have really liked to just return a plain `TaskReport` or better yet a generic `T extends TaskReport` from the `OverlordResourceTestClient` and the caller would handle it correctly e.g. `MsqTaskReport`. In that case, you wouldn't even need this class.
But it's not feasible since we return a `Map<String, TaskReport>` which is a weird thing since the
method is `getTaskReport()`. I suppose we could encapsulate all of it in a single wrapper class,
but that can come later.
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/AbstractTestQueryHelper.java:
##########
@@ -146,24 +147,29 @@ private void testQueries(String url, List<QueryResultType> queries) throws Excep
for (QueryResultType queryWithResult : queries) {
LOG.info("Running Query %s", queryWithResult.getQuery());
List<Map<String, Object>> result = queryClient.query(url, queryWithResult.getQuery());
- if (!QueryResultVerifier.compareResults(result, queryWithResult.getExpectedResults(),
- queryWithResult.getFieldsToTest()
- )) {
+ Optional<String> resultsComparison = QueryResultVerifier.compareResults(result, queryWithResult.getExpectedResults(),
+ queryWithResult.getFieldsToTest());
Review Comment:
Nit: indentation seems off
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/msq/ITMultiStageQuery.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testsEx.msq;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.inject.Inject;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.CoordinatorResourceTestClient;
+import org.apache.druid.testing.clients.MsqTestClient;
+import org.apache.druid.testing.utils.DataLoaderHelper;
+import org.apache.druid.testing.utils.MsqTestQueryHelper;
+import org.apache.druid.testsEx.categories.MultiStageQuery;
+import org.apache.druid.testsEx.config.DruidTestRunner;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+@RunWith(DruidTestRunner.class)
+@Category(MultiStageQuery.class)
+public class ITMultiStageQuery
+{
+ @Inject
+ private MsqTestQueryHelper msqHelper;
+
+ @Inject
+ private MsqTestClient msqClient;
+
+ @Inject
+ private IntegrationTestingConfig config;
+
+ @Inject
+ private ObjectMapper jsonMapper;
+
+ @Inject
+ private DataLoaderHelper dataLoaderHelper;
+
+ @Inject
+ private CoordinatorResourceTestClient coordinatorClient;
+
+ private static final String QUERY_FILE = "/indexer/wikipedia_index_data1_query.json";
+
+ @Test
+ public void testMsqIngestionAndQuerying() throws Exception
+ {
+ String datasource = "dst";
+
+ // Clear up the datasource from the previous runs
+ coordinatorClient.unloadSegmentsForDataSource(datasource);
+
+ String queryLocal =
+ StringUtils.format(
+ "INSERT INTO %s\n"
Review Comment:
Sure, we can do it later.
But the reason to move it into a dedicated file is not the size of the query, rather the readability (you can avoid the escapes, newlines and never-ending pluses 😃 ) and homogeneity. We shouldn't mix the patterns i.e. write some queries in files and some in the code.
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/QueryResultVerifier.java:
##########
@@ -19,13 +19,19 @@
package org.apache.druid.testing.utils;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.logger.Logger;
+
import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
public class QueryResultVerifier
{
- public static boolean compareResults(
+ private static final Logger LOG = new Logger(QueryResultVerifier.class);
+
+ public static Optional<String> compareResults(
Review Comment:
Having an Optional of error message is better than just the boolean. But the best thing to do in such cases is to have a dedicated response object, which contains a success status and an error message. But since this is for tests only, we can do it later.
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/QueryResultVerifier.java:
##########
@@ -41,19 +47,35 @@ public static boolean compareResults(
if (fieldsToTest != null && !fieldsToTest.isEmpty()) {
for (String field : fieldsToTest) {
if (!actualRes.get(field).equals(expRes.get(field))) {
- return false;
+ String mismatchMessage = StringUtils.format(
+ "Field [%s] mismatch. Expected: [%s], Actual: [%s]",
Review Comment:
Something like this would be more informative:
```suggestion
"Mismatch in row [%d], col [%s]. Expected: [%s], Actual: [%s]",
```
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/QueryResultVerifier.java:
##########
@@ -25,12 +25,13 @@
import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
public class QueryResultVerifier
{
private static final Logger LOG = new Logger(QueryResultVerifier.class);
- public static boolean compareResults(
+ public static Optional<String> compareResults(
Review Comment:
+1
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] LakshSingla commented on a diff in pull request #12992: Add IT for MSQ task engine using the new IT framework
Posted by GitBox <gi...@apache.org>.
LakshSingla commented on code in PR #12992:
URL: https://github.com/apache/druid/pull/12992#discussion_r968138951
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/MsqTestQueryHelper.java:
##########
@@ -0,0 +1,236 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.utils;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import org.apache.druid.indexer.TaskState;
+import org.apache.druid.indexer.TaskStatusPlus;
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.guava.Yielder;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.msq.indexing.report.MSQResultsReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReportPayload;
+import org.apache.druid.msq.sql.SqlTaskStatus;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.sql.http.SqlQuery;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.MsqOverlordResourceTestClient;
+import org.apache.druid.testing.clients.MsqTestClient;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+/**
+ * Helper class to aid out ITs for MSQ
+ */
+public class MsqTestQueryHelper extends AbstractTestQueryHelper<MsqQueryWithResults>
+{
+
+ private final ObjectMapper jsonMapper;
+ private final IntegrationTestingConfig config;
+ private final MsqOverlordResourceTestClient overlordClient;
+ private final MsqTestClient msqClient;
+
+
+ @Inject
+ MsqTestQueryHelper(
+ final ObjectMapper jsonMapper,
+ final MsqTestClient queryClient,
+ final IntegrationTestingConfig config,
+ final MsqOverlordResourceTestClient overlordClient,
+ final MsqTestClient msqClient
+ )
+ {
+ super(jsonMapper, queryClient, config);
+ this.jsonMapper = jsonMapper;
+ this.config = config;
+ this.overlordClient = overlordClient;
+ this.msqClient = msqClient;
+ }
+
+ @Override
+ public String getQueryURL(String schemeAndHost)
+ {
+ return StringUtils.format("%s/druid/v2/sql/task", schemeAndHost);
+ }
+
+ /**
+ * Submits a task to the MSQ API with the given query string, and default headers and parameters
+ */
+ public String submitMsqTask(String sqlQueryString) throws ExecutionException, InterruptedException
+ {
+ return submitMsqTask(new SqlQuery(sqlQueryString, null, false, false, false, ImmutableMap.of(), null));
+ }
+
+ // Run the task, wait for it to complete, fetch the reports, verify the results,
+
+ /**
+ * Submits a {@link SqlQuery} to the MSQ API for execution. This method waits for the task to be accepted by the cluster
+ * and returns the task id associated with the submitted task
+ */
+ public String submitMsqTask(SqlQuery sqlQuery) throws ExecutionException, InterruptedException
+ {
+ String queryUrl = getQueryURL(config.getBrokerUrl());
+ Future<StatusResponseHolder> responseHolderFuture = msqClient.queryAsync(queryUrl, sqlQuery);
+ // It is okay to block here for the result because MSQ tasks return the task Id associated with it, which shouldn't
+ // consume a lot of time
+ StatusResponseHolder statusResponseHolder = responseHolderFuture.get();
+
+ // Check if the task has been accepted successfully
+ HttpResponseStatus httpResponseStatus = statusResponseHolder.getStatus();
+ if (!httpResponseStatus.equals(HttpResponseStatus.ACCEPTED)) {
+ throw new ISE(
+ "Unable to submit the task successfully. Received response status code [%d], and response content:\n[%s]",
+ httpResponseStatus,
+ statusResponseHolder.getContent()
+ );
+ }
+ String content = statusResponseHolder.getContent();
+ SqlTaskStatus sqlTaskStatus;
+ try {
+ sqlTaskStatus = jsonMapper.readValue(content, SqlTaskStatus.class);
+ }
+ catch (JsonProcessingException e) {
+ throw new ISE("Unable to parse the response");
+ }
+ if (sqlTaskStatus.getState().isFailure()) {
+ throw new ISE("Unable to start the task successfully.\nPossible exception: %s", sqlTaskStatus.getError());
+ }
+ return sqlTaskStatus.getTaskId();
+ }
+
+ /**
+ * Polls the overlord API every 1 second and waits for a submitted MSQ task to be completed. Alternatively, one can
+ * specify the maximum time to poll. The method returns the last fetched {@link TaskState} of the task
+ */
+ public TaskState pollTaskIdForCompletion(String taskId, long maxTimeoutSeconds)
+ {
+ if (maxTimeoutSeconds < 0) {
+ throw new IAE("Timeout cannot be negative");
+ } else if (maxTimeoutSeconds == 0) {
+ maxTimeoutSeconds = Long.MAX_VALUE;
+ }
+ long time = 0;
+ do {
+ TaskStatusPlus taskStatusPlus = overlordClient.getTaskStatus(taskId);
+ TaskState statusCode = taskStatusPlus.getStatusCode();
+ if (statusCode != null && statusCode.isComplete()) {
+ return taskStatusPlus.getStatusCode();
+ }
+ try {
+ Thread.sleep(1000);
+ }
+ catch (InterruptedException e) {
+ throw new ISE(e, "Interrupted while polling for task [%s] completion", taskId);
+ }
+ } while (time++ < maxTimeoutSeconds);
+ return overlordClient.getTaskStatus(taskId).getStatusCode();
+ }
+
+ /**
+ * Fetches status reports for a given task
+ */
+ public Map<String, MSQTaskReport> fetchStatusReports(String taskId)
+ {
+ return overlordClient.getTaskReportForMsqTask(taskId);
+ }
+
+ /**
+ * Compares the results for a given taskId. It is required that the task has produced some results that can be verified
+ */
+ public void compareResults(String taskId, MsqQueryWithResults expectedQueryWithResults)
+ {
+ Map<String, MSQTaskReport> statusReport = fetchStatusReports(taskId);
+ MSQTaskReport taskReport = statusReport.get(MSQTaskReport.REPORT_KEY);
+ if (taskReport == null) {
+ throw new ISE("Unable to fetch the status report for the task [%]", taskId);
+ }
+ MSQTaskReportPayload taskReportPayload = Preconditions.checkNotNull(
+ taskReport.getPayload(),
+ "payload"
+ );
+ MSQResultsReport resultsReport = Preconditions.checkNotNull(
+ taskReportPayload.getResults(),
+ "Results report for the task id is empty"
+ );
+
+ List<Map<String, Object>> actualResults = new ArrayList<>();
+
+ Yielder<Object[]> yielder = resultsReport.getResultYielder();
+ RowSignature rowSignature = resultsReport.getSignature();
+
+ while (!yielder.isDone()) {
+ Object[] row = yielder.get();
+ Map<String, Object> rowWithFieldNames = new LinkedHashMap<>();
+ for (int i = 0; i < row.length; ++i) {
+ rowWithFieldNames.put(rowSignature.getColumnName(i), row[i]);
+ }
+ actualResults.add(rowWithFieldNames);
+ yielder = yielder.next(null);
+ }
+
+ boolean resultsComparison = QueryResultVerifier.compareResults(
+ actualResults,
+ expectedQueryWithResults.getExpectedResults(),
+ Collections.emptyList()
+ );
+ if (!resultsComparison) {
+ throw new IAE("Expected query result is different from the actual result");
Review Comment:
I tried to refactor this but ITQueryRetryTestOnMissingSegments uses the pass/fail value to determine whether to fail the tests or not. Therefore failing in the helper method might not be correct. I am logging the mismatched values/size for now in the QueryResultsVerifier.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] LakshSingla commented on a diff in pull request #12992: Add IT for MSQ task engine using the new IT framework
Posted by GitBox <gi...@apache.org>.
LakshSingla commented on code in PR #12992:
URL: https://github.com/apache/druid/pull/12992#discussion_r968140037
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/msq/ITMultiStageQuery.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testsEx.msq;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.inject.Inject;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.CoordinatorResourceTestClient;
+import org.apache.druid.testing.clients.MsqTestClient;
+import org.apache.druid.testing.utils.DataLoaderHelper;
+import org.apache.druid.testing.utils.MsqTestQueryHelper;
+import org.apache.druid.testsEx.categories.MultiStageQuery;
+import org.apache.druid.testsEx.config.DruidTestRunner;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+@RunWith(DruidTestRunner.class)
+@Category(MultiStageQuery.class)
+public class ITMultiStageQuery
+{
+ @Inject
+ private MsqTestQueryHelper msqHelper;
+
+ @Inject
+ private MsqTestClient msqClient;
+
+ @Inject
+ private IntegrationTestingConfig config;
+
+ @Inject
+ private ObjectMapper jsonMapper;
+
+ @Inject
+ private DataLoaderHelper dataLoaderHelper;
+
+ @Inject
+ private CoordinatorResourceTestClient coordinatorClient;
+
+ private static final String QUERY_FILE = "/indexer/wikipedia_index_data1_query.json";
+
+ @Test
+ public void testMsqIngestionAndQuerying() throws Exception
+ {
+ String datasource = "dst";
+
+ // Clear up the datasource from the previous runs
+ coordinatorClient.unloadSegmentsForDataSource(datasource);
+
+ String queryLocal =
+ StringUtils.format(
+ "INSERT INTO %s\n"
Review Comment:
Deferring this change because SQL queries are generally supposed to be small enough to be written in the code. If we have a lot of similar INSERT queries with different data sources, I think we can extract them out in a separate resource file.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] LakshSingla commented on a diff in pull request #12992: Add IT for MSQ task engine using the new IT framework
Posted by GitBox <gi...@apache.org>.
LakshSingla commented on code in PR #12992:
URL: https://github.com/apache/druid/pull/12992#discussion_r973825160
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/QueryResultVerifier.java:
##########
@@ -19,13 +19,19 @@
package org.apache.druid.testing.utils;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.logger.Logger;
+
import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
public class QueryResultVerifier
{
- public static boolean compareResults(
+ private static final Logger LOG = new Logger(QueryResultVerifier.class);
+
+ public static Optional<String> compareResults(
Review Comment:
I would defer against the suggestion of `QueryResultVerifier.assertEqual(...);` because the helper is not going to do assertions based on the result. This is because the same method is utilized in `ITQueryRetryTestOnMissingSegments.java` which depending on a flag fails or drops the mismatch in the rows. Therefore one way would be to create a separate method that helps perform the Assertions. But since this is not required in several places, I haven't created a such method.
Good point regarding the test case in the planner. That would certainly be a cleaner and easier approach to add the test cases.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] cryptoe commented on a diff in pull request #12992: Add IT for MSQ task engine using the new IT framework
Posted by GitBox <gi...@apache.org>.
cryptoe commented on code in PR #12992:
URL: https://github.com/apache/druid/pull/12992#discussion_r972668683
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/QueryResultVerifier.java:
##########
@@ -25,12 +25,13 @@
import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
public class QueryResultVerifier
{
private static final Logger LOG = new Logger(QueryResultVerifier.class);
- public static boolean compareResults(
+ public static Optional<String> compareResults(
Review Comment:
Nit: As I see this method being used a lot, lets javadoc this
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] LakshSingla commented on a diff in pull request #12992: Add IT for MSQ task engine using the new IT framework
Posted by GitBox <gi...@apache.org>.
LakshSingla commented on code in PR #12992:
URL: https://github.com/apache/druid/pull/12992#discussion_r956912355
##########
integration-tests/src/main/java/org/apache/druid/testing/clients/MsqTestClient.java:
##########
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.clients;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.inject.Inject;
+import org.apache.druid.java.util.http.client.HttpClient;
+import org.apache.druid.sql.http.SqlQuery;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.guice.TestClient;
+
+import javax.ws.rs.core.MediaType;
+
+public class MsqTestClient extends AbstractQueryResourceTestClient<SqlQuery>
Review Comment:
`SqlResourceTestClient` implements the same AbstractQueryResourceTestClient without any modification therefore I wanted to maintain a parallel b/w them. Also, the type parameter passed in the QueryResourceTestClient and the MsqTestClient are different.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] LakshSingla commented on a diff in pull request #12992: Add IT for MSQ task engine using the new IT framework
Posted by GitBox <gi...@apache.org>.
LakshSingla commented on code in PR #12992:
URL: https://github.com/apache/druid/pull/12992#discussion_r973810074
##########
integration-tests-ex/cases/cluster/Common/environment-configs/common.env:
##########
@@ -49,7 +49,7 @@ DRUID_INSTANCE=
# variables: druid_standard_loadList defined here, and druid_test_loadList, defined
# in a docker-compose.yaml file, for any test-specific extensions.
# See compose.md for more details.
-druid_standard_loadList=mysql-metadata-storage,it-tools,druid-lookups-cached-global,druid-histogram,druid-datasketches,druid-parquet-extensions,druid-avro-extensions,druid-protobuf-extensions,druid-orc-extensions,druid-kafka-indexing-service,druid-s3-extensions
+druid_standard_loadList=mysql-metadata-storage,it-tools,druid-lookups-cached-global,druid-histogram,druid-datasketches,druid-parquet-extensions,druid-avro-extensions,druid-protobuf-extensions,druid-orc-extensions,druid-kafka-indexing-service,druid-s3-extensions,druid-multi-stage-query
Review Comment:
The extension would be loaded for every integration test setup. There is a way to load it just for MSQ tests, but I made the change in the common env because the druid configs (quickstart one) contained the druid-multi-stage-query by default.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] LakshSingla commented on a diff in pull request #12992: Add IT for MSQ task engine using the new IT framework
Posted by GitBox <gi...@apache.org>.
LakshSingla commented on code in PR #12992:
URL: https://github.com/apache/druid/pull/12992#discussion_r973828416
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/MsqTestQueryHelper.java:
##########
@@ -0,0 +1,246 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.utils;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import org.apache.druid.indexer.TaskState;
+import org.apache.druid.indexer.TaskStatusPlus;
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.RetryUtils;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.guava.Yielder;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.msq.indexing.report.MSQResultsReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReportPayload;
+import org.apache.druid.msq.sql.SqlTaskStatus;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.sql.http.SqlQuery;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.SqlResourceTestClient;
+import org.apache.druid.testing.clients.msq.MsqOverlordResourceTestClient;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+/**
+ * Helper class to aid out ITs for MSQ.
+ * This takes all the clients required to make the necessary calls to the client APIs for MSQ and performs the boilerplate
+ * functions for the tests
+ */
+public class MsqTestQueryHelper extends AbstractTestQueryHelper<MsqQueryWithResults>
+{
+
+ private final ObjectMapper jsonMapper;
+ private final IntegrationTestingConfig config;
+ private final MsqOverlordResourceTestClient overlordClient;
+ private final SqlResourceTestClient msqClient;
+
+
+ @Inject
+ MsqTestQueryHelper(
+ final ObjectMapper jsonMapper,
+ final SqlResourceTestClient queryClient,
+ final IntegrationTestingConfig config,
+ final MsqOverlordResourceTestClient overlordClient,
+ final SqlResourceTestClient msqClient
+ )
+ {
+ super(jsonMapper, queryClient, config);
+ this.jsonMapper = jsonMapper;
+ this.config = config;
+ this.overlordClient = overlordClient;
+ this.msqClient = msqClient;
+ }
+
+ @Override
+ public String getQueryURL(String schemeAndHost)
+ {
+ return StringUtils.format("%s/druid/v2/sql/task", schemeAndHost);
+ }
+
+ /**
+ * Submits a task to the MSQ API with the given query string, and default headers and parameters
+ */
+ public String submitMsqTask(String sqlQueryString) throws ExecutionException, InterruptedException
+ {
+ return submitMsqTask(new SqlQuery(sqlQueryString, null, false, false, false, ImmutableMap.of(), null));
+ }
+
+ // Run the task, wait for it to complete, fetch the reports, verify the results,
+
+ /**
+ * Submits a {@link SqlQuery} to the MSQ API for execution. This method waits for the task to be accepted by the cluster
+ * and returns the task id associated with the submitted task
+ */
+ public String submitMsqTask(SqlQuery sqlQuery) throws ExecutionException, InterruptedException
+ {
+ String queryUrl = getQueryURL(config.getBrokerUrl());
+ Future<StatusResponseHolder> responseHolderFuture = msqClient.queryAsync(queryUrl, sqlQuery);
+ // It is okay to block here for the result because MSQ tasks return the task Id associated with it, which shouldn't
+ // consume a lot of time
+ StatusResponseHolder statusResponseHolder = responseHolderFuture.get();
+
+ // Check if the task has been accepted successfully
+ HttpResponseStatus httpResponseStatus = statusResponseHolder.getStatus();
+ if (!httpResponseStatus.equals(HttpResponseStatus.ACCEPTED)) {
+ throw new ISE(
+ "Unable to submit the task successfully. Received response status code [%d], and response content:\n[%s]",
+ httpResponseStatus,
+ statusResponseHolder.getContent()
+ );
+ }
+ String content = statusResponseHolder.getContent();
+ SqlTaskStatus sqlTaskStatus;
+ try {
+ sqlTaskStatus = jsonMapper.readValue(content, SqlTaskStatus.class);
+ }
+ catch (JsonProcessingException e) {
+ throw new ISE("Unable to parse the response");
+ }
+ if (sqlTaskStatus.getState().isFailure()) {
Review Comment:
Good point. Updated in the tests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] kfaraz commented on a diff in pull request #12992: Add IT for MSQ task engine using the new IT framework
Posted by GitBox <gi...@apache.org>.
kfaraz commented on code in PR #12992:
URL: https://github.com/apache/druid/pull/12992#discussion_r972794810
##########
integration-tests/src/main/java/org/apache/druid/testing/clients/msq/MsqOverlordResourceTestClient.java:
##########
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.clients.msq;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.inject.Inject;
+import org.apache.druid.guice.annotations.Json;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.http.client.HttpClient;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.msq.guice.MSQIndexingModule;
+import org.apache.druid.msq.indexing.report.MSQTaskReport;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.OverlordResourceTestClient;
+import org.apache.druid.testing.guice.TestClient;
+import org.jboss.netty.handler.codec.http.HttpMethod;
+
+import java.util.Map;
+
+/**
+ * Overlord resource client for MSQ Tasks
+ */
+public class MsqOverlordResourceTestClient extends OverlordResourceTestClient
Review Comment:
These MSQ specific test utilities such as `MsqOverlordResourceTestClient`, `MsqQueryWithResults`, `MsqTestQueryHelper` will be used for MSQ ITs only. So they should live in the same place as the `ITMultiStageQuery` itself.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] kfaraz commented on a diff in pull request #12992: Add IT for MSQ task engine using the new IT framework
Posted by GitBox <gi...@apache.org>.
kfaraz commented on code in PR #12992:
URL: https://github.com/apache/druid/pull/12992#discussion_r977181796
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/MsqTestQueryHelper.java:
##########
@@ -0,0 +1,246 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.utils;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import org.apache.druid.indexer.TaskState;
+import org.apache.druid.indexer.TaskStatusPlus;
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.RetryUtils;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.guava.Yielder;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.msq.indexing.report.MSQResultsReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReportPayload;
+import org.apache.druid.msq.sql.SqlTaskStatus;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.sql.http.SqlQuery;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.SqlResourceTestClient;
+import org.apache.druid.testing.clients.msq.MsqOverlordResourceTestClient;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+/**
+ * Helper class to aid out ITs for MSQ.
+ * This takes all the clients required to make the necessary calls to the client APIs for MSQ and performs the boilerplate
+ * functions for the tests
+ */
+public class MsqTestQueryHelper extends AbstractTestQueryHelper<MsqQueryWithResults>
+{
+
+ private final ObjectMapper jsonMapper;
+ private final IntegrationTestingConfig config;
+ private final MsqOverlordResourceTestClient overlordClient;
+ private final SqlResourceTestClient msqClient;
+
+
+ @Inject
+ MsqTestQueryHelper(
+ final ObjectMapper jsonMapper,
+ final SqlResourceTestClient queryClient,
+ final IntegrationTestingConfig config,
+ final MsqOverlordResourceTestClient overlordClient,
+ final SqlResourceTestClient msqClient
+ )
+ {
+ super(jsonMapper, queryClient, config);
+ this.jsonMapper = jsonMapper;
+ this.config = config;
+ this.overlordClient = overlordClient;
+ this.msqClient = msqClient;
+ }
+
+ @Override
+ public String getQueryURL(String schemeAndHost)
+ {
+ return StringUtils.format("%s/druid/v2/sql/task", schemeAndHost);
+ }
+
+ /**
+ * Submits a task to the MSQ API with the given query string, and default headers and parameters
+ */
+ public String submitMsqTask(String sqlQueryString) throws ExecutionException, InterruptedException
+ {
+ return submitMsqTask(new SqlQuery(sqlQueryString, null, false, false, false, ImmutableMap.of(), null));
+ }
+
+ // Run the task, wait for it to complete, fetch the reports, verify the results,
+
+ /**
+ * Submits a {@link SqlQuery} to the MSQ API for execution. This method waits for the task to be accepted by the cluster
+ * and returns the task id associated with the submitted task
+ */
+ public String submitMsqTask(SqlQuery sqlQuery) throws ExecutionException, InterruptedException
+ {
+ String queryUrl = getQueryURL(config.getBrokerUrl());
+ Future<StatusResponseHolder> responseHolderFuture = msqClient.queryAsync(queryUrl, sqlQuery);
+ // It is okay to block here for the result because MSQ tasks return the task Id associated with it, which shouldn't
+ // consume a lot of time
+ StatusResponseHolder statusResponseHolder = responseHolderFuture.get();
+
+ // Check if the task has been accepted successfully
+ HttpResponseStatus httpResponseStatus = statusResponseHolder.getStatus();
+ if (!httpResponseStatus.equals(HttpResponseStatus.ACCEPTED)) {
+ throw new ISE(
+ "Unable to submit the task successfully. Received response status code [%d], and response content:\n[%s]",
+ httpResponseStatus,
+ statusResponseHolder.getContent()
+ );
+ }
+ String content = statusResponseHolder.getContent();
+ SqlTaskStatus sqlTaskStatus;
+ try {
+ sqlTaskStatus = jsonMapper.readValue(content, SqlTaskStatus.class);
+ }
+ catch (JsonProcessingException e) {
+ throw new ISE("Unable to parse the response");
+ }
+ if (sqlTaskStatus.getState().isFailure()) {
+ throw new ISE("Unable to start the task successfully.\nPossible exception: %s", sqlTaskStatus.getError());
+ }
+ return sqlTaskStatus.getTaskId();
+ }
+
+ /**
+ * The method retries till the task with taskId gets completed i.e. {@link TaskState#isComplete()}} returns true and
+ * returns the last fetched state {@link TaskState} of the task
+ */
+ public TaskState pollTaskIdForCompletion(String taskId) throws Exception
+ {
+ return RetryUtils.retry(
+ () -> {
+ TaskStatusPlus taskStatusPlus = overlordClient.getTaskStatus(taskId);
+ TaskState statusCode = taskStatusPlus.getStatusCode();
+ if (statusCode != null && statusCode.isComplete()) {
+ return taskStatusPlus.getStatusCode();
+ }
+ throw new TaskStillRunningException();
+ },
+ (Throwable t) -> t instanceof TaskStillRunningException,
+ 100
+ );
+ }
+
+ /**
+ * Fetches status reports for a given task
+ */
+ public Map<String, MSQTaskReport> fetchStatusReports(String taskId)
Review Comment:
Yeah, we could just return the `MSQTaskReport` here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] cryptoe commented on a diff in pull request #12992: Add IT for MSQ task engine using the new IT framework
Posted by GitBox <gi...@apache.org>.
cryptoe commented on code in PR #12992:
URL: https://github.com/apache/druid/pull/12992#discussion_r970367807
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/MsqTestQueryHelper.java:
##########
@@ -0,0 +1,236 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.utils;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import org.apache.druid.indexer.TaskState;
+import org.apache.druid.indexer.TaskStatusPlus;
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.guava.Yielder;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.msq.indexing.report.MSQResultsReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReportPayload;
+import org.apache.druid.msq.sql.SqlTaskStatus;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.sql.http.SqlQuery;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.MsqOverlordResourceTestClient;
+import org.apache.druid.testing.clients.MsqTestClient;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+/**
+ * Helper class to aid out ITs for MSQ
+ */
+public class MsqTestQueryHelper extends AbstractTestQueryHelper<MsqQueryWithResults>
+{
+
+ private final ObjectMapper jsonMapper;
+ private final IntegrationTestingConfig config;
+ private final MsqOverlordResourceTestClient overlordClient;
+ private final MsqTestClient msqClient;
+
+
+ @Inject
+ MsqTestQueryHelper(
+ final ObjectMapper jsonMapper,
+ final MsqTestClient queryClient,
+ final IntegrationTestingConfig config,
+ final MsqOverlordResourceTestClient overlordClient,
+ final MsqTestClient msqClient
+ )
+ {
+ super(jsonMapper, queryClient, config);
+ this.jsonMapper = jsonMapper;
+ this.config = config;
+ this.overlordClient = overlordClient;
+ this.msqClient = msqClient;
+ }
+
+ @Override
+ public String getQueryURL(String schemeAndHost)
+ {
+ return StringUtils.format("%s/druid/v2/sql/task", schemeAndHost);
+ }
+
+ /**
+ * Submits a task to the MSQ API with the given query string, and default headers and parameters
+ */
+ public String submitMsqTask(String sqlQueryString) throws ExecutionException, InterruptedException
+ {
+ return submitMsqTask(new SqlQuery(sqlQueryString, null, false, false, false, ImmutableMap.of(), null));
+ }
+
+ // Run the task, wait for it to complete, fetch the reports, verify the results,
+
+ /**
+ * Submits a {@link SqlQuery} to the MSQ API for execution. This method waits for the task to be accepted by the cluster
+ * and returns the task id associated with the submitted task
+ */
+ public String submitMsqTask(SqlQuery sqlQuery) throws ExecutionException, InterruptedException
+ {
+ String queryUrl = getQueryURL(config.getBrokerUrl());
+ Future<StatusResponseHolder> responseHolderFuture = msqClient.queryAsync(queryUrl, sqlQuery);
+ // It is okay to block here for the result because MSQ tasks return the task Id associated with it, which shouldn't
+ // consume a lot of time
+ StatusResponseHolder statusResponseHolder = responseHolderFuture.get();
+
+ // Check if the task has been accepted successfully
+ HttpResponseStatus httpResponseStatus = statusResponseHolder.getStatus();
+ if (!httpResponseStatus.equals(HttpResponseStatus.ACCEPTED)) {
+ throw new ISE(
+ "Unable to submit the task successfully. Received response status code [%d], and response content:\n[%s]",
+ httpResponseStatus,
+ statusResponseHolder.getContent()
+ );
+ }
+ String content = statusResponseHolder.getContent();
+ SqlTaskStatus sqlTaskStatus;
+ try {
+ sqlTaskStatus = jsonMapper.readValue(content, SqlTaskStatus.class);
+ }
+ catch (JsonProcessingException e) {
+ throw new ISE("Unable to parse the response");
+ }
+ if (sqlTaskStatus.getState().isFailure()) {
+ throw new ISE("Unable to start the task successfully.\nPossible exception: %s", sqlTaskStatus.getError());
+ }
+ return sqlTaskStatus.getTaskId();
+ }
+
+ /**
+ * Polls the overlord API every 1 second and waits for a submitted MSQ task to be completed. Alternatively, one can
+ * specify the maximum time to poll. The method returns the last fetched {@link TaskState} of the task
+ */
+ public TaskState pollTaskIdForCompletion(String taskId, long maxTimeoutSeconds)
+ {
+ if (maxTimeoutSeconds < 0) {
+ throw new IAE("Timeout cannot be negative");
+ } else if (maxTimeoutSeconds == 0) {
+ maxTimeoutSeconds = Long.MAX_VALUE;
+ }
+ long time = 0;
+ do {
+ TaskStatusPlus taskStatusPlus = overlordClient.getTaskStatus(taskId);
+ TaskState statusCode = taskStatusPlus.getStatusCode();
+ if (statusCode != null && statusCode.isComplete()) {
+ return taskStatusPlus.getStatusCode();
+ }
+ try {
+ Thread.sleep(1000);
+ }
+ catch (InterruptedException e) {
+ throw new ISE(e, "Interrupted while polling for task [%s] completion", taskId);
+ }
+ } while (time++ < maxTimeoutSeconds);
+ return overlordClient.getTaskStatus(taskId).getStatusCode();
+ }
+
+ /**
+ * Fetches status reports for a given task
+ */
+ public Map<String, MSQTaskReport> fetchStatusReports(String taskId)
+ {
+ return overlordClient.getTaskReportForMsqTask(taskId);
+ }
+
+ /**
+ * Compares the results for a given taskId. It is required that the task has produced some results that can be verified
+ */
+ public void compareResults(String taskId, MsqQueryWithResults expectedQueryWithResults)
+ {
+ Map<String, MSQTaskReport> statusReport = fetchStatusReports(taskId);
+ MSQTaskReport taskReport = statusReport.get(MSQTaskReport.REPORT_KEY);
+ if (taskReport == null) {
+ throw new ISE("Unable to fetch the status report for the task [%]", taskId);
+ }
+ MSQTaskReportPayload taskReportPayload = Preconditions.checkNotNull(
+ taskReport.getPayload(),
+ "payload"
+ );
+ MSQResultsReport resultsReport = Preconditions.checkNotNull(
+ taskReportPayload.getResults(),
+ "Results report for the task id is empty"
+ );
+
+ List<Map<String, Object>> actualResults = new ArrayList<>();
+
+ Yielder<Object[]> yielder = resultsReport.getResultYielder();
+ RowSignature rowSignature = resultsReport.getSignature();
+
+ while (!yielder.isDone()) {
+ Object[] row = yielder.get();
+ Map<String, Object> rowWithFieldNames = new LinkedHashMap<>();
+ for (int i = 0; i < row.length; ++i) {
+ rowWithFieldNames.put(rowSignature.getColumnName(i), row[i]);
+ }
+ actualResults.add(rowWithFieldNames);
+ yielder = yielder.next(null);
+ }
+
+ boolean resultsComparison = QueryResultVerifier.compareResults(
+ actualResults,
+ expectedQueryWithResults.getExpectedResults(),
+ Collections.emptyList()
+ );
+ if (!resultsComparison) {
+ throw new IAE("Expected query result is different from the actual result");
Review Comment:
I second Kashif's opinion. The signature of the resultVerifier can be "Optional<StringErrorMessage>" and then the caller can do whatever it wants.
Though one version in the results verifier should always throw an exception.
void resultsEqualsOrThrow()
Something like that.
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/MsqTestQueryHelper.java:
##########
@@ -0,0 +1,235 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.utils;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import org.apache.druid.indexer.TaskState;
+import org.apache.druid.indexer.TaskStatusPlus;
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.RetryUtils;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.guava.Yielder;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.msq.indexing.report.MSQResultsReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReportPayload;
+import org.apache.druid.msq.sql.SqlTaskStatus;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.sql.http.SqlQuery;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.SqlResourceTestClient;
+import org.apache.druid.testing.clients.msq.MsqOverlordResourceTestClient;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+/**
+ * Helper class to aid out ITs for MSQ.
+ * This takes all the clients required to make the necessary calls to the client APIs for MSQ and performs the boiler
+ * plate functions for the tests
+ */
+public class MsqTestQueryHelper extends AbstractTestQueryHelper<MsqQueryWithResults>
+{
+
+ private final ObjectMapper jsonMapper;
+ private final IntegrationTestingConfig config;
+ private final MsqOverlordResourceTestClient overlordClient;
+ private final SqlResourceTestClient msqClient;
+
+
+ @Inject
+ MsqTestQueryHelper(
+ final ObjectMapper jsonMapper,
+ final SqlResourceTestClient queryClient,
+ final IntegrationTestingConfig config,
+ final MsqOverlordResourceTestClient overlordClient,
+ final SqlResourceTestClient msqClient
+ )
+ {
+ super(jsonMapper, queryClient, config);
+ this.jsonMapper = jsonMapper;
+ this.config = config;
+ this.overlordClient = overlordClient;
+ this.msqClient = msqClient;
+ }
+
+ @Override
+ public String getQueryURL(String schemeAndHost)
+ {
+ return StringUtils.format("%s/druid/v2/sql/task", schemeAndHost);
+ }
+
+ /**
+ * Submits a task to the MSQ API with the given query string, and default headers and parameters
+ */
+ public String submitMsqTask(String sqlQueryString) throws ExecutionException, InterruptedException
+ {
+ return submitMsqTask(new SqlQuery(sqlQueryString, null, false, false, false, ImmutableMap.of(), null));
+ }
+
+ // Run the task, wait for it to complete, fetch the reports, verify the results,
+
+ /**
+ * Submits a {@link SqlQuery} to the MSQ API for execution. This method waits for the task to be accepted by the cluster
+ * and returns the task id associated with the submitted task
+ */
+ public String submitMsqTask(SqlQuery sqlQuery) throws ExecutionException, InterruptedException
+ {
+ String queryUrl = getQueryURL(config.getBrokerUrl());
+ Future<StatusResponseHolder> responseHolderFuture = msqClient.queryAsync(queryUrl, sqlQuery);
+ // It is okay to block here for the result because MSQ tasks return the task Id associated with it, which shouldn't
+ // consume a lot of time
+ StatusResponseHolder statusResponseHolder = responseHolderFuture.get();
+
+ // Check if the task has been accepted successfully
+ HttpResponseStatus httpResponseStatus = statusResponseHolder.getStatus();
+ if (!httpResponseStatus.equals(HttpResponseStatus.ACCEPTED)) {
+ throw new ISE(
+ "Unable to submit the task successfully. Received response status code [%d], and response content:\n[%s]",
+ httpResponseStatus,
+ statusResponseHolder.getContent()
+ );
+ }
+ String content = statusResponseHolder.getContent();
+ SqlTaskStatus sqlTaskStatus;
+ try {
+ sqlTaskStatus = jsonMapper.readValue(content, SqlTaskStatus.class);
+ }
+ catch (JsonProcessingException e) {
+ throw new ISE("Unable to parse the response");
+ }
+ if (sqlTaskStatus.getState().isFailure()) {
+ throw new ISE("Unable to start the task successfully.\nPossible exception: %s", sqlTaskStatus.getError());
+ }
+ return sqlTaskStatus.getTaskId();
+ }
+
+ /**
+ * Polls the overlord API every 1 second and waits for a submitted MSQ task to be completed. Alternatively, one can
+ * specify the maximum time to poll. The method returns the last fetched {@link TaskState} of the task
Review Comment:
Nit: Javadocs might need to update here.
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/QueryResultVerifier.java:
##########
@@ -41,17 +46,29 @@ public static boolean compareResults(
if (fieldsToTest != null && !fieldsToTest.isEmpty()) {
for (String field : fieldsToTest) {
if (!actualRes.get(field).equals(expRes.get(field))) {
+ LOG.info(StringUtils.format(
+ "Field [%s] mismatch. Expected: [%s], Actual: [%s]",
+ field,
+ expRes,
+ actualRes
+ ));
return false;
}
}
} else {
if (!actualRes.equals(expRes)) {
+ LOG.info(StringUtils.format(
+ "Row mismatch. Expected: [%s], Actual: [%s]",
+ expRes,
+ actualRes
+ ));
return false;
}
}
}
if (actualIter.hasNext() || expectedIter.hasNext()) {
+ LOG.info("Results size mismatch");
Review Comment:
Please mention if there are fewer results or more results if possible.
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/QueryResultVerifier.java:
##########
@@ -19,12 +19,17 @@
package org.apache.druid.testing.utils;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.logger.Logger;
+
import java.util.Iterator;
import java.util.List;
import java.util.Map;
public class QueryResultVerifier
{
+ private static final Logger LOG = new Logger(QueryResultVerifier.class);
+
public static boolean compareResults(
Review Comment:
I think this method should throw an exception when there is a mismatch else every caller would have to do that handling.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] LakshSingla commented on a diff in pull request #12992: Add IT for MSQ task engine using the new IT framework
Posted by GitBox <gi...@apache.org>.
LakshSingla commented on code in PR #12992:
URL: https://github.com/apache/druid/pull/12992#discussion_r958020962
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/MsqTestQueryHelper.java:
##########
@@ -0,0 +1,236 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.utils;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import org.apache.druid.indexer.TaskState;
+import org.apache.druid.indexer.TaskStatusPlus;
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.guava.Yielder;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.msq.indexing.report.MSQResultsReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReportPayload;
+import org.apache.druid.msq.sql.SqlTaskStatus;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.sql.http.SqlQuery;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.MsqOverlordResourceTestClient;
+import org.apache.druid.testing.clients.MsqTestClient;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+/**
+ * Helper class to aid out ITs for MSQ
+ */
+public class MsqTestQueryHelper extends AbstractTestQueryHelper<MsqQueryWithResults>
Review Comment:
Removing the `MsqTestClient` class, reusing `SqlResourceTestClient` instead of it as per @cryptoe's suggestion in one of the above comments, and retaining the helper methods here. IMO helper class should utilize the client class to perform the function that the unit test requires. I will update the Javadoc to make it clearer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] LakshSingla commented on a diff in pull request #12992: Add IT for MSQ task engine using the new IT framework
Posted by GitBox <gi...@apache.org>.
LakshSingla commented on code in PR #12992:
URL: https://github.com/apache/druid/pull/12992#discussion_r958019700
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/MsqTestQueryHelper.java:
##########
@@ -0,0 +1,236 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.utils;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import org.apache.druid.indexer.TaskState;
+import org.apache.druid.indexer.TaskStatusPlus;
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.guava.Yielder;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.msq.indexing.report.MSQResultsReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReportPayload;
+import org.apache.druid.msq.sql.SqlTaskStatus;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.sql.http.SqlQuery;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.MsqOverlordResourceTestClient;
+import org.apache.druid.testing.clients.MsqTestClient;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+/**
+ * Helper class to aid out ITs for MSQ
+ */
+public class MsqTestQueryHelper extends AbstractTestQueryHelper<MsqQueryWithResults>
+{
+
+ private final ObjectMapper jsonMapper;
+ private final IntegrationTestingConfig config;
+ private final MsqOverlordResourceTestClient overlordClient;
+ private final MsqTestClient msqClient;
+
+
+ @Inject
+ MsqTestQueryHelper(
+ final ObjectMapper jsonMapper,
+ final MsqTestClient queryClient,
+ final IntegrationTestingConfig config,
+ final MsqOverlordResourceTestClient overlordClient,
+ final MsqTestClient msqClient
+ )
+ {
+ super(jsonMapper, queryClient, config);
+ this.jsonMapper = jsonMapper;
+ this.config = config;
+ this.overlordClient = overlordClient;
+ this.msqClient = msqClient;
+ }
+
+ @Override
+ public String getQueryURL(String schemeAndHost)
+ {
+ return StringUtils.format("%s/druid/v2/sql/task", schemeAndHost);
+ }
+
+ /**
+ * Submits a task to the MSQ API with the given query string, and default headers and parameters
+ */
+ public String submitMsqTask(String sqlQueryString) throws ExecutionException, InterruptedException
+ {
+ return submitMsqTask(new SqlQuery(sqlQueryString, null, false, false, false, ImmutableMap.of(), null));
+ }
+
+ // Run the task, wait for it to complete, fetch the reports, verify the results,
+
+ /**
+ * Submits a {@link SqlQuery} to the MSQ API for execution. This method waits for the task to be accepted by the cluster
+ * and returns the task id associated with the submitted task
+ */
+ public String submitMsqTask(SqlQuery sqlQuery) throws ExecutionException, InterruptedException
+ {
+ String queryUrl = getQueryURL(config.getBrokerUrl());
+ Future<StatusResponseHolder> responseHolderFuture = msqClient.queryAsync(queryUrl, sqlQuery);
+ // It is okay to block here for the result because MSQ tasks return the task Id associated with it, which shouldn't
+ // consume a lot of time
+ StatusResponseHolder statusResponseHolder = responseHolderFuture.get();
+
+ // Check if the task has been accepted successfully
+ HttpResponseStatus httpResponseStatus = statusResponseHolder.getStatus();
+ if (!httpResponseStatus.equals(HttpResponseStatus.ACCEPTED)) {
+ throw new ISE(
+ "Unable to submit the task successfully. Received response status code [%d], and response content:\n[%s]",
+ httpResponseStatus,
+ statusResponseHolder.getContent()
+ );
+ }
+ String content = statusResponseHolder.getContent();
+ SqlTaskStatus sqlTaskStatus;
+ try {
+ sqlTaskStatus = jsonMapper.readValue(content, SqlTaskStatus.class);
+ }
+ catch (JsonProcessingException e) {
+ throw new ISE("Unable to parse the response");
+ }
+ if (sqlTaskStatus.getState().isFailure()) {
+ throw new ISE("Unable to start the task successfully.\nPossible exception: %s", sqlTaskStatus.getError());
+ }
+ return sqlTaskStatus.getTaskId();
+ }
+
+ /**
+ * Polls the overlord API every 1 second and waits for a submitted MSQ task to be completed. Alternatively, one can
+ * specify the maximum time to poll. The method returns the last fetched {@link TaskState} of the task
+ */
+ public TaskState pollTaskIdForCompletion(String taskId, long maxTimeoutSeconds)
+ {
+ if (maxTimeoutSeconds < 0) {
+ throw new IAE("Timeout cannot be negative");
+ } else if (maxTimeoutSeconds == 0) {
+ maxTimeoutSeconds = Long.MAX_VALUE;
+ }
+ long time = 0;
+ do {
+ TaskStatusPlus taskStatusPlus = overlordClient.getTaskStatus(taskId);
+ TaskState statusCode = taskStatusPlus.getStatusCode();
+ if (statusCode != null && statusCode.isComplete()) {
+ return taskStatusPlus.getStatusCode();
+ }
+ try {
+ Thread.sleep(1000);
+ }
+ catch (InterruptedException e) {
+ throw new ISE(e, "Interrupted while polling for task [%s] completion", taskId);
+ }
+ } while (time++ < maxTimeoutSeconds);
Review Comment:
Since we are not retrying but polling for the task state, and the exponential backoff nature of the retries in the RetryUtil, I was trying to avoid using the class. I have rewritten the code using RetryUtils, and it does seem to be cleaner.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] kfaraz commented on a diff in pull request #12992: Add IT for MSQ task engine using the new IT framework
Posted by GitBox <gi...@apache.org>.
kfaraz commented on code in PR #12992:
URL: https://github.com/apache/druid/pull/12992#discussion_r957105315
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/MsqTestQueryHelper.java:
##########
@@ -0,0 +1,236 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.utils;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import org.apache.druid.indexer.TaskState;
+import org.apache.druid.indexer.TaskStatusPlus;
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.guava.Yielder;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.msq.indexing.report.MSQResultsReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReportPayload;
+import org.apache.druid.msq.sql.SqlTaskStatus;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.sql.http.SqlQuery;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.MsqOverlordResourceTestClient;
+import org.apache.druid.testing.clients.MsqTestClient;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+/**
+ * Helper class to aid out ITs for MSQ
+ */
+public class MsqTestQueryHelper extends AbstractTestQueryHelper<MsqQueryWithResults>
+{
+
+ private final ObjectMapper jsonMapper;
+ private final IntegrationTestingConfig config;
+ private final MsqOverlordResourceTestClient overlordClient;
+ private final MsqTestClient msqClient;
+
+
+ @Inject
+ MsqTestQueryHelper(
+ final ObjectMapper jsonMapper,
+ final MsqTestClient queryClient,
+ final IntegrationTestingConfig config,
+ final MsqOverlordResourceTestClient overlordClient,
+ final MsqTestClient msqClient
+ )
+ {
+ super(jsonMapper, queryClient, config);
+ this.jsonMapper = jsonMapper;
+ this.config = config;
+ this.overlordClient = overlordClient;
+ this.msqClient = msqClient;
+ }
+
+ @Override
+ public String getQueryURL(String schemeAndHost)
+ {
+ return StringUtils.format("%s/druid/v2/sql/task", schemeAndHost);
+ }
+
+ /**
+ * Submits a task to the MSQ API with the given query string, and default headers and parameters
+ */
+ public String submitMsqTask(String sqlQueryString) throws ExecutionException, InterruptedException
+ {
+ return submitMsqTask(new SqlQuery(sqlQueryString, null, false, false, false, ImmutableMap.of(), null));
+ }
+
+ // Run the task, wait for it to complete, fetch the reports, verify the results,
+
+ /**
+ * Submits a {@link SqlQuery} to the MSQ API for execution. This method waits for the task to be accepted by the cluster
+ * and returns the task id associated with the submitted task
+ */
+ public String submitMsqTask(SqlQuery sqlQuery) throws ExecutionException, InterruptedException
+ {
+ String queryUrl = getQueryURL(config.getBrokerUrl());
+ Future<StatusResponseHolder> responseHolderFuture = msqClient.queryAsync(queryUrl, sqlQuery);
+ // It is okay to block here for the result because MSQ tasks return the task Id associated with it, which shouldn't
+ // consume a lot of time
+ StatusResponseHolder statusResponseHolder = responseHolderFuture.get();
+
+ // Check if the task has been accepted successfully
+ HttpResponseStatus httpResponseStatus = statusResponseHolder.getStatus();
+ if (!httpResponseStatus.equals(HttpResponseStatus.ACCEPTED)) {
+ throw new ISE(
+ "Unable to submit the task successfully. Received response status code [%d], and response content:\n[%s]",
+ httpResponseStatus,
+ statusResponseHolder.getContent()
+ );
+ }
+ String content = statusResponseHolder.getContent();
+ SqlTaskStatus sqlTaskStatus;
+ try {
+ sqlTaskStatus = jsonMapper.readValue(content, SqlTaskStatus.class);
+ }
+ catch (JsonProcessingException e) {
+ throw new ISE("Unable to parse the response");
+ }
+ if (sqlTaskStatus.getState().isFailure()) {
+ throw new ISE("Unable to start the task successfully.\nPossible exception: %s", sqlTaskStatus.getError());
+ }
+ return sqlTaskStatus.getTaskId();
+ }
+
+ /**
+ * Polls the overlord API every 1 second and waits for a submitted MSQ task to be completed. Alternatively, one can
+ * specify the maximum time to poll. The method returns the last fetched {@link TaskState} of the task
+ */
+ public TaskState pollTaskIdForCompletion(String taskId, long maxTimeoutSeconds)
+ {
+ if (maxTimeoutSeconds < 0) {
+ throw new IAE("Timeout cannot be negative");
+ } else if (maxTimeoutSeconds == 0) {
+ maxTimeoutSeconds = Long.MAX_VALUE;
+ }
+ long time = 0;
+ do {
+ TaskStatusPlus taskStatusPlus = overlordClient.getTaskStatus(taskId);
+ TaskState statusCode = taskStatusPlus.getStatusCode();
+ if (statusCode != null && statusCode.isComplete()) {
+ return taskStatusPlus.getStatusCode();
+ }
+ try {
+ Thread.sleep(1000);
+ }
+ catch (InterruptedException e) {
+ throw new ISE(e, "Interrupted while polling for task [%s] completion", taskId);
+ }
+ } while (time++ < maxTimeoutSeconds);
+ return overlordClient.getTaskStatus(taskId).getStatusCode();
+ }
+
+ /**
+ * Fetches status reports for a given task
+ */
+ public Map<String, MSQTaskReport> fetchStatusReports(String taskId)
+ {
+ return overlordClient.getTaskReportForMsqTask(taskId);
+ }
+
+ /**
+ * Compares the results for a given taskId. It is required that the task has produced some results that can be verified
+ */
+ public void compareResults(String taskId, MsqQueryWithResults expectedQueryWithResults)
Review Comment:
This method should not be public.
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/msq/ITMultiStageQuery.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testsEx.msq;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.inject.Inject;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.CoordinatorResourceTestClient;
+import org.apache.druid.testing.clients.MsqTestClient;
+import org.apache.druid.testing.utils.DataLoaderHelper;
+import org.apache.druid.testing.utils.MsqTestQueryHelper;
+import org.apache.druid.testsEx.categories.MultiStageQuery;
+import org.apache.druid.testsEx.config.DruidTestRunner;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+@RunWith(DruidTestRunner.class)
+@Category(MultiStageQuery.class)
+public class ITMultiStageQuery
+{
+ @Inject
+ private MsqTestQueryHelper msqHelper;
+
+ @Inject
+ private MsqTestClient msqClient;
+
+ @Inject
+ private IntegrationTestingConfig config;
+
+ @Inject
+ private ObjectMapper jsonMapper;
+
+ @Inject
+ private DataLoaderHelper dataLoaderHelper;
+
+ @Inject
+ private CoordinatorResourceTestClient coordinatorClient;
+
+ private static final String QUERY_FILE = "/indexer/wikipedia_index_data1_query.json";
+
+ @Test
+ public void testMsqIngestionAndQuerying() throws Exception
+ {
+ String datasource = "dst";
+
+ // Clear up the datasource from the previous runs
+ coordinatorClient.unloadSegmentsForDataSource(datasource);
+
+ String queryLocal =
+ StringUtils.format(
+ "INSERT INTO %s\n"
Review Comment:
I suppose it would be cleaner to have the INSERT query in a separate resource file, similar to the SELECT query.
##########
integration-tests/src/main/java/org/apache/druid/testing/clients/MsqOverlordResourceTestClient.java:
##########
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.clients;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableList;
+import com.google.inject.Inject;
+import org.apache.druid.guice.annotations.Json;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.http.client.HttpClient;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.msq.guice.MSQIndexingModule;
+import org.apache.druid.msq.indexing.report.MSQTaskReport;
+import org.apache.druid.segment.virtual.ExpressionVirtualColumn;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.guice.TestClient;
+import org.jboss.netty.handler.codec.http.HttpMethod;
+
+import java.util.Map;
+
+/**
+ * Overlord resource client for MSQ Tasks
+ */
+public class MsqOverlordResourceTestClient extends OverlordResourceTestClient
Review Comment:
This and other related classes should probably be in the `msq` package as they will only ever be used for MSQ ITs.
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/MsqTestQueryHelper.java:
##########
@@ -0,0 +1,236 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.utils;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import org.apache.druid.indexer.TaskState;
+import org.apache.druid.indexer.TaskStatusPlus;
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.guava.Yielder;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.msq.indexing.report.MSQResultsReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReportPayload;
+import org.apache.druid.msq.sql.SqlTaskStatus;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.sql.http.SqlQuery;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.MsqOverlordResourceTestClient;
+import org.apache.druid.testing.clients.MsqTestClient;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+/**
+ * Helper class to aid out ITs for MSQ
+ */
+public class MsqTestQueryHelper extends AbstractTestQueryHelper<MsqQueryWithResults>
Review Comment:
Design wise, the contract of this class is not very clear. If it is just meant to be a query helper, it should not be submitting tasks and fetching reports. Maybe that functionality should be retained by the `MsqTestClient` itself.
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/MsqTestQueryHelper.java:
##########
@@ -0,0 +1,236 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.utils;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import org.apache.druid.indexer.TaskState;
+import org.apache.druid.indexer.TaskStatusPlus;
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.guava.Yielder;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.msq.indexing.report.MSQResultsReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReportPayload;
+import org.apache.druid.msq.sql.SqlTaskStatus;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.sql.http.SqlQuery;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.MsqOverlordResourceTestClient;
+import org.apache.druid.testing.clients.MsqTestClient;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+/**
+ * Helper class to aid out ITs for MSQ
+ */
+public class MsqTestQueryHelper extends AbstractTestQueryHelper<MsqQueryWithResults>
+{
+
+ private final ObjectMapper jsonMapper;
+ private final IntegrationTestingConfig config;
+ private final MsqOverlordResourceTestClient overlordClient;
+ private final MsqTestClient msqClient;
+
+
+ @Inject
+ MsqTestQueryHelper(
+ final ObjectMapper jsonMapper,
+ final MsqTestClient queryClient,
+ final IntegrationTestingConfig config,
+ final MsqOverlordResourceTestClient overlordClient,
+ final MsqTestClient msqClient
+ )
+ {
+ super(jsonMapper, queryClient, config);
+ this.jsonMapper = jsonMapper;
+ this.config = config;
+ this.overlordClient = overlordClient;
+ this.msqClient = msqClient;
+ }
+
+ @Override
+ public String getQueryURL(String schemeAndHost)
+ {
+ return StringUtils.format("%s/druid/v2/sql/task", schemeAndHost);
+ }
+
+ /**
+ * Submits a task to the MSQ API with the given query string, and default headers and parameters
+ */
+ public String submitMsqTask(String sqlQueryString) throws ExecutionException, InterruptedException
+ {
+ return submitMsqTask(new SqlQuery(sqlQueryString, null, false, false, false, ImmutableMap.of(), null));
+ }
+
+ // Run the task, wait for it to complete, fetch the reports, verify the results,
+
+ /**
+ * Submits a {@link SqlQuery} to the MSQ API for execution. This method waits for the task to be accepted by the cluster
+ * and returns the task id associated with the submitted task
+ */
+ public String submitMsqTask(SqlQuery sqlQuery) throws ExecutionException, InterruptedException
+ {
+ String queryUrl = getQueryURL(config.getBrokerUrl());
+ Future<StatusResponseHolder> responseHolderFuture = msqClient.queryAsync(queryUrl, sqlQuery);
+ // It is okay to block here for the result because MSQ tasks return the task Id associated with it, which shouldn't
+ // consume a lot of time
+ StatusResponseHolder statusResponseHolder = responseHolderFuture.get();
+
+ // Check if the task has been accepted successfully
+ HttpResponseStatus httpResponseStatus = statusResponseHolder.getStatus();
+ if (!httpResponseStatus.equals(HttpResponseStatus.ACCEPTED)) {
+ throw new ISE(
+ "Unable to submit the task successfully. Received response status code [%d], and response content:\n[%s]",
+ httpResponseStatus,
+ statusResponseHolder.getContent()
+ );
+ }
+ String content = statusResponseHolder.getContent();
+ SqlTaskStatus sqlTaskStatus;
+ try {
+ sqlTaskStatus = jsonMapper.readValue(content, SqlTaskStatus.class);
+ }
+ catch (JsonProcessingException e) {
+ throw new ISE("Unable to parse the response");
+ }
+ if (sqlTaskStatus.getState().isFailure()) {
+ throw new ISE("Unable to start the task successfully.\nPossible exception: %s", sqlTaskStatus.getError());
+ }
+ return sqlTaskStatus.getTaskId();
+ }
+
+ /**
+ * Polls the overlord API every 1 second and waits for a submitted MSQ task to be completed. Alternatively, one can
+ * specify the maximum time to poll. The method returns the last fetched {@link TaskState} of the task
+ */
+ public TaskState pollTaskIdForCompletion(String taskId, long maxTimeoutSeconds)
+ {
+ if (maxTimeoutSeconds < 0) {
+ throw new IAE("Timeout cannot be negative");
+ } else if (maxTimeoutSeconds == 0) {
+ maxTimeoutSeconds = Long.MAX_VALUE;
+ }
+ long time = 0;
+ do {
+ TaskStatusPlus taskStatusPlus = overlordClient.getTaskStatus(taskId);
+ TaskState statusCode = taskStatusPlus.getStatusCode();
+ if (statusCode != null && statusCode.isComplete()) {
+ return taskStatusPlus.getStatusCode();
+ }
+ try {
+ Thread.sleep(1000);
+ }
+ catch (InterruptedException e) {
+ throw new ISE(e, "Interrupted while polling for task [%s] completion", taskId);
+ }
+ } while (time++ < maxTimeoutSeconds);
+ return overlordClient.getTaskStatus(taskId).getStatusCode();
+ }
+
+ /**
+ * Fetches status reports for a given task
+ */
+ public Map<String, MSQTaskReport> fetchStatusReports(String taskId)
+ {
+ return overlordClient.getTaskReportForMsqTask(taskId);
+ }
+
+ /**
+ * Compares the results for a given taskId. It is required that the task has produced some results that can be verified
+ */
+ public void compareResults(String taskId, MsqQueryWithResults expectedQueryWithResults)
+ {
+ Map<String, MSQTaskReport> statusReport = fetchStatusReports(taskId);
+ MSQTaskReport taskReport = statusReport.get(MSQTaskReport.REPORT_KEY);
+ if (taskReport == null) {
+ throw new ISE("Unable to fetch the status report for the task [%]", taskId);
+ }
+ MSQTaskReportPayload taskReportPayload = Preconditions.checkNotNull(
+ taskReport.getPayload(),
+ "payload"
+ );
+ MSQResultsReport resultsReport = Preconditions.checkNotNull(
+ taskReportPayload.getResults(),
+ "Results report for the task id is empty"
+ );
+
+ List<Map<String, Object>> actualResults = new ArrayList<>();
+
+ Yielder<Object[]> yielder = resultsReport.getResultYielder();
+ RowSignature rowSignature = resultsReport.getSignature();
+
+ while (!yielder.isDone()) {
+ Object[] row = yielder.get();
+ Map<String, Object> rowWithFieldNames = new LinkedHashMap<>();
+ for (int i = 0; i < row.length; ++i) {
+ rowWithFieldNames.put(rowSignature.getColumnName(i), row[i]);
+ }
+ actualResults.add(rowWithFieldNames);
+ yielder = yielder.next(null);
+ }
+
+ boolean resultsComparison = QueryResultVerifier.compareResults(
+ actualResults,
+ expectedQueryWithResults.getExpectedResults(),
+ Collections.emptyList()
+ );
+ if (!resultsComparison) {
+ throw new IAE("Expected query result is different from the actual result");
Review Comment:
Maybe `QueryResultVerifier` itself should do the assertions or atleast give back messages detailing the mismatch. The caller getting a boolean and then throwing an ISE makes for very difficult to debug tests.
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/MsqTestQueryHelper.java:
##########
@@ -0,0 +1,236 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.utils;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import org.apache.druid.indexer.TaskState;
+import org.apache.druid.indexer.TaskStatusPlus;
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.guava.Yielder;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.msq.indexing.report.MSQResultsReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReportPayload;
+import org.apache.druid.msq.sql.SqlTaskStatus;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.sql.http.SqlQuery;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.MsqOverlordResourceTestClient;
+import org.apache.druid.testing.clients.MsqTestClient;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+/**
+ * Helper class to aid out ITs for MSQ
+ */
+public class MsqTestQueryHelper extends AbstractTestQueryHelper<MsqQueryWithResults>
+{
+
+ private final ObjectMapper jsonMapper;
+ private final IntegrationTestingConfig config;
+ private final MsqOverlordResourceTestClient overlordClient;
+ private final MsqTestClient msqClient;
+
+
+ @Inject
+ MsqTestQueryHelper(
+ final ObjectMapper jsonMapper,
+ final MsqTestClient queryClient,
+ final IntegrationTestingConfig config,
+ final MsqOverlordResourceTestClient overlordClient,
+ final MsqTestClient msqClient
+ )
+ {
+ super(jsonMapper, queryClient, config);
+ this.jsonMapper = jsonMapper;
+ this.config = config;
+ this.overlordClient = overlordClient;
+ this.msqClient = msqClient;
+ }
+
+ @Override
+ public String getQueryURL(String schemeAndHost)
+ {
+ return StringUtils.format("%s/druid/v2/sql/task", schemeAndHost);
+ }
+
+ /**
+ * Submits a task to the MSQ API with the given query string, and default headers and parameters
+ */
+ public String submitMsqTask(String sqlQueryString) throws ExecutionException, InterruptedException
+ {
+ return submitMsqTask(new SqlQuery(sqlQueryString, null, false, false, false, ImmutableMap.of(), null));
+ }
+
+ // Run the task, wait for it to complete, fetch the reports, verify the results,
+
+ /**
+ * Submits a {@link SqlQuery} to the MSQ API for execution. This method waits for the task to be accepted by the cluster
+ * and returns the task id associated with the submitted task
+ */
+ public String submitMsqTask(SqlQuery sqlQuery) throws ExecutionException, InterruptedException
+ {
+ String queryUrl = getQueryURL(config.getBrokerUrl());
+ Future<StatusResponseHolder> responseHolderFuture = msqClient.queryAsync(queryUrl, sqlQuery);
+ // It is okay to block here for the result because MSQ tasks return the task Id associated with it, which shouldn't
+ // consume a lot of time
+ StatusResponseHolder statusResponseHolder = responseHolderFuture.get();
+
+ // Check if the task has been accepted successfully
+ HttpResponseStatus httpResponseStatus = statusResponseHolder.getStatus();
+ if (!httpResponseStatus.equals(HttpResponseStatus.ACCEPTED)) {
+ throw new ISE(
+ "Unable to submit the task successfully. Received response status code [%d], and response content:\n[%s]",
+ httpResponseStatus,
+ statusResponseHolder.getContent()
+ );
+ }
+ String content = statusResponseHolder.getContent();
+ SqlTaskStatus sqlTaskStatus;
+ try {
+ sqlTaskStatus = jsonMapper.readValue(content, SqlTaskStatus.class);
+ }
+ catch (JsonProcessingException e) {
+ throw new ISE("Unable to parse the response");
+ }
+ if (sqlTaskStatus.getState().isFailure()) {
+ throw new ISE("Unable to start the task successfully.\nPossible exception: %s", sqlTaskStatus.getError());
+ }
+ return sqlTaskStatus.getTaskId();
+ }
+
+ /**
+ * Polls the overlord API every 1 second and waits for a submitted MSQ task to be completed. Alternatively, one can
+ * specify the maximum time to poll. The method returns the last fetched {@link TaskState} of the task
+ */
+ public TaskState pollTaskIdForCompletion(String taskId, long maxTimeoutSeconds)
+ {
+ if (maxTimeoutSeconds < 0) {
+ throw new IAE("Timeout cannot be negative");
+ } else if (maxTimeoutSeconds == 0) {
+ maxTimeoutSeconds = Long.MAX_VALUE;
+ }
+ long time = 0;
+ do {
+ TaskStatusPlus taskStatusPlus = overlordClient.getTaskStatus(taskId);
+ TaskState statusCode = taskStatusPlus.getStatusCode();
+ if (statusCode != null && statusCode.isComplete()) {
+ return taskStatusPlus.getStatusCode();
+ }
+ try {
+ Thread.sleep(1000);
+ }
+ catch (InterruptedException e) {
+ throw new ISE(e, "Interrupted while polling for task [%s] completion", taskId);
+ }
+ } while (time++ < maxTimeoutSeconds);
+ return overlordClient.getTaskStatus(taskId).getStatusCode();
+ }
+
+ /**
+ * Fetches status reports for a given task
+ */
+ public Map<String, MSQTaskReport> fetchStatusReports(String taskId)
+ {
+ return overlordClient.getTaskReportForMsqTask(taskId);
+ }
+
+ /**
+ * Compares the results for a given taskId. It is required that the task has produced some results that can be verified
+ */
+ public void compareResults(String taskId, MsqQueryWithResults expectedQueryWithResults)
+ {
+ Map<String, MSQTaskReport> statusReport = fetchStatusReports(taskId);
+ MSQTaskReport taskReport = statusReport.get(MSQTaskReport.REPORT_KEY);
+ if (taskReport == null) {
+ throw new ISE("Unable to fetch the status report for the task [%]", taskId);
+ }
+ MSQTaskReportPayload taskReportPayload = Preconditions.checkNotNull(
+ taskReport.getPayload(),
+ "payload"
+ );
+ MSQResultsReport resultsReport = Preconditions.checkNotNull(
+ taskReportPayload.getResults(),
+ "Results report for the task id is empty"
+ );
+
+ List<Map<String, Object>> actualResults = new ArrayList<>();
+
+ Yielder<Object[]> yielder = resultsReport.getResultYielder();
+ RowSignature rowSignature = resultsReport.getSignature();
+
+ while (!yielder.isDone()) {
+ Object[] row = yielder.get();
+ Map<String, Object> rowWithFieldNames = new LinkedHashMap<>();
+ for (int i = 0; i < row.length; ++i) {
+ rowWithFieldNames.put(rowSignature.getColumnName(i), row[i]);
+ }
+ actualResults.add(rowWithFieldNames);
+ yielder = yielder.next(null);
+ }
+
+ boolean resultsComparison = QueryResultVerifier.compareResults(
+ actualResults,
+ expectedQueryWithResults.getExpectedResults(),
+ Collections.emptyList()
+ );
+ if (!resultsComparison) {
+ throw new IAE("Expected query result is different from the actual result");
+ }
+ }
+
+ /**
+ * Runs queries from files using MSQ and compares the results with the ones provided
+ */
+ public void testQueriesFromFileUsingMsq(String filePath, String fullDatasourcePath)
Review Comment:
The `UsingMsq` suffix is probably redundant as this class is `MsqTestQueryHelper` and will only be used with MSQ. You need not worry about clashing with the method in the parent class, you can always override it.
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/msq/ITMultiStageQuery.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testsEx.msq;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.inject.Inject;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.CoordinatorResourceTestClient;
+import org.apache.druid.testing.clients.MsqTestClient;
+import org.apache.druid.testing.utils.DataLoaderHelper;
+import org.apache.druid.testing.utils.MsqTestQueryHelper;
+import org.apache.druid.testsEx.categories.MultiStageQuery;
+import org.apache.druid.testsEx.config.DruidTestRunner;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+@RunWith(DruidTestRunner.class)
+@Category(MultiStageQuery.class)
+public class ITMultiStageQuery
+{
+ @Inject
+ private MsqTestQueryHelper msqHelper;
+
+ @Inject
+ private MsqTestClient msqClient;
+
+ @Inject
+ private IntegrationTestingConfig config;
+
+ @Inject
+ private ObjectMapper jsonMapper;
+
+ @Inject
+ private DataLoaderHelper dataLoaderHelper;
+
+ @Inject
+ private CoordinatorResourceTestClient coordinatorClient;
+
+ private static final String QUERY_FILE = "/indexer/wikipedia_index_data1_query.json";
Review Comment:
I think we should use more specific names for the resource files that we add. Over time, these file names become difficult to tell apart and we often end up duplicating the content.
You could either name them based on the tests they are used in or the function they perform (e.g. select_orderby_xyz_query.json). It would also be nice if they were kept in folders that correspondeded to the test package.
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/MsqTestQueryHelper.java:
##########
@@ -0,0 +1,236 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.utils;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import org.apache.druid.indexer.TaskState;
+import org.apache.druid.indexer.TaskStatusPlus;
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.guava.Yielder;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.msq.indexing.report.MSQResultsReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReportPayload;
+import org.apache.druid.msq.sql.SqlTaskStatus;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.sql.http.SqlQuery;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.MsqOverlordResourceTestClient;
+import org.apache.druid.testing.clients.MsqTestClient;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+/**
+ * Helper class to aid out ITs for MSQ
+ */
+public class MsqTestQueryHelper extends AbstractTestQueryHelper<MsqQueryWithResults>
+{
+
+ private final ObjectMapper jsonMapper;
+ private final IntegrationTestingConfig config;
+ private final MsqOverlordResourceTestClient overlordClient;
+ private final MsqTestClient msqClient;
+
+
+ @Inject
+ MsqTestQueryHelper(
+ final ObjectMapper jsonMapper,
+ final MsqTestClient queryClient,
+ final IntegrationTestingConfig config,
+ final MsqOverlordResourceTestClient overlordClient,
+ final MsqTestClient msqClient
+ )
+ {
+ super(jsonMapper, queryClient, config);
+ this.jsonMapper = jsonMapper;
+ this.config = config;
+ this.overlordClient = overlordClient;
+ this.msqClient = msqClient;
+ }
+
+ @Override
+ public String getQueryURL(String schemeAndHost)
+ {
+ return StringUtils.format("%s/druid/v2/sql/task", schemeAndHost);
+ }
+
+ /**
+ * Submits a task to the MSQ API with the given query string, and default headers and parameters
+ */
+ public String submitMsqTask(String sqlQueryString) throws ExecutionException, InterruptedException
+ {
+ return submitMsqTask(new SqlQuery(sqlQueryString, null, false, false, false, ImmutableMap.of(), null));
+ }
+
+ // Run the task, wait for it to complete, fetch the reports, verify the results,
+
+ /**
+ * Submits a {@link SqlQuery} to the MSQ API for execution. This method waits for the task to be accepted by the cluster
+ * and returns the task id associated with the submitted task
+ */
+ public String submitMsqTask(SqlQuery sqlQuery) throws ExecutionException, InterruptedException
+ {
+ String queryUrl = getQueryURL(config.getBrokerUrl());
+ Future<StatusResponseHolder> responseHolderFuture = msqClient.queryAsync(queryUrl, sqlQuery);
+ // It is okay to block here for the result because MSQ tasks return the task Id associated with it, which shouldn't
+ // consume a lot of time
+ StatusResponseHolder statusResponseHolder = responseHolderFuture.get();
+
+ // Check if the task has been accepted successfully
+ HttpResponseStatus httpResponseStatus = statusResponseHolder.getStatus();
+ if (!httpResponseStatus.equals(HttpResponseStatus.ACCEPTED)) {
+ throw new ISE(
+ "Unable to submit the task successfully. Received response status code [%d], and response content:\n[%s]",
+ httpResponseStatus,
+ statusResponseHolder.getContent()
+ );
+ }
+ String content = statusResponseHolder.getContent();
+ SqlTaskStatus sqlTaskStatus;
+ try {
+ sqlTaskStatus = jsonMapper.readValue(content, SqlTaskStatus.class);
+ }
+ catch (JsonProcessingException e) {
+ throw new ISE("Unable to parse the response");
+ }
+ if (sqlTaskStatus.getState().isFailure()) {
+ throw new ISE("Unable to start the task successfully.\nPossible exception: %s", sqlTaskStatus.getError());
+ }
+ return sqlTaskStatus.getTaskId();
+ }
+
+ /**
+ * Polls the overlord API every 1 second and waits for a submitted MSQ task to be completed. Alternatively, one can
+ * specify the maximum time to poll. The method returns the last fetched {@link TaskState} of the task
+ */
+ public TaskState pollTaskIdForCompletion(String taskId, long maxTimeoutSeconds)
+ {
+ if (maxTimeoutSeconds < 0) {
+ throw new IAE("Timeout cannot be negative");
+ } else if (maxTimeoutSeconds == 0) {
+ maxTimeoutSeconds = Long.MAX_VALUE;
+ }
+ long time = 0;
+ do {
+ TaskStatusPlus taskStatusPlus = overlordClient.getTaskStatus(taskId);
+ TaskState statusCode = taskStatusPlus.getStatusCode();
+ if (statusCode != null && statusCode.isComplete()) {
+ return taskStatusPlus.getStatusCode();
+ }
+ try {
+ Thread.sleep(1000);
+ }
+ catch (InterruptedException e) {
+ throw new ISE(e, "Interrupted while polling for task [%s] completion", taskId);
+ }
+ } while (time++ < maxTimeoutSeconds);
+ return overlordClient.getTaskStatus(taskId).getStatusCode();
+ }
+
+ /**
+ * Fetches status reports for a given task
+ */
+ public Map<String, MSQTaskReport> fetchStatusReports(String taskId)
+ {
+ return overlordClient.getTaskReportForMsqTask(taskId);
+ }
+
+ /**
+ * Compares the results for a given taskId. It is required that the task has produced some results that can be verified
+ */
+ public void compareResults(String taskId, MsqQueryWithResults expectedQueryWithResults)
+ {
+ Map<String, MSQTaskReport> statusReport = fetchStatusReports(taskId);
+ MSQTaskReport taskReport = statusReport.get(MSQTaskReport.REPORT_KEY);
+ if (taskReport == null) {
+ throw new ISE("Unable to fetch the status report for the task [%]", taskId);
+ }
+ MSQTaskReportPayload taskReportPayload = Preconditions.checkNotNull(
+ taskReport.getPayload(),
+ "payload"
+ );
+ MSQResultsReport resultsReport = Preconditions.checkNotNull(
+ taskReportPayload.getResults(),
+ "Results report for the task id is empty"
+ );
+
+ List<Map<String, Object>> actualResults = new ArrayList<>();
+
+ Yielder<Object[]> yielder = resultsReport.getResultYielder();
+ RowSignature rowSignature = resultsReport.getSignature();
+
+ while (!yielder.isDone()) {
+ Object[] row = yielder.get();
+ Map<String, Object> rowWithFieldNames = new LinkedHashMap<>();
+ for (int i = 0; i < row.length; ++i) {
+ rowWithFieldNames.put(rowSignature.getColumnName(i), row[i]);
+ }
+ actualResults.add(rowWithFieldNames);
+ yielder = yielder.next(null);
+ }
+
+ boolean resultsComparison = QueryResultVerifier.compareResults(
+ actualResults,
+ expectedQueryWithResults.getExpectedResults(),
+ Collections.emptyList()
+ );
+ if (!resultsComparison) {
+ throw new IAE("Expected query result is different from the actual result");
Review Comment:
I know that `AbstractTestQueryHelper` also does this in a similar way but we should avoid this if we want to write cleaner tests. Assertions should preferably be done using the appropriate JUnit/TestNG constructs.
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/MsqTestQueryHelper.java:
##########
@@ -0,0 +1,236 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.utils;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import org.apache.druid.indexer.TaskState;
+import org.apache.druid.indexer.TaskStatusPlus;
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.guava.Yielder;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.msq.indexing.report.MSQResultsReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReportPayload;
+import org.apache.druid.msq.sql.SqlTaskStatus;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.sql.http.SqlQuery;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.MsqOverlordResourceTestClient;
+import org.apache.druid.testing.clients.MsqTestClient;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+/**
+ * Helper class to aid out ITs for MSQ
+ */
+public class MsqTestQueryHelper extends AbstractTestQueryHelper<MsqQueryWithResults>
+{
+
+ private final ObjectMapper jsonMapper;
+ private final IntegrationTestingConfig config;
+ private final MsqOverlordResourceTestClient overlordClient;
+ private final MsqTestClient msqClient;
+
+
+ @Inject
+ MsqTestQueryHelper(
+ final ObjectMapper jsonMapper,
+ final MsqTestClient queryClient,
+ final IntegrationTestingConfig config,
+ final MsqOverlordResourceTestClient overlordClient,
+ final MsqTestClient msqClient
+ )
+ {
+ super(jsonMapper, queryClient, config);
+ this.jsonMapper = jsonMapper;
+ this.config = config;
+ this.overlordClient = overlordClient;
+ this.msqClient = msqClient;
+ }
+
+ @Override
+ public String getQueryURL(String schemeAndHost)
+ {
+ return StringUtils.format("%s/druid/v2/sql/task", schemeAndHost);
+ }
+
+ /**
+ * Submits a task to the MSQ API with the given query string, and default headers and parameters
+ */
+ public String submitMsqTask(String sqlQueryString) throws ExecutionException, InterruptedException
+ {
+ return submitMsqTask(new SqlQuery(sqlQueryString, null, false, false, false, ImmutableMap.of(), null));
+ }
+
+ // Run the task, wait for it to complete, fetch the reports, verify the results,
+
+ /**
+ * Submits a {@link SqlQuery} to the MSQ API for execution. This method waits for the task to be accepted by the cluster
+ * and returns the task id associated with the submitted task
+ */
+ public String submitMsqTask(SqlQuery sqlQuery) throws ExecutionException, InterruptedException
+ {
+ String queryUrl = getQueryURL(config.getBrokerUrl());
+ Future<StatusResponseHolder> responseHolderFuture = msqClient.queryAsync(queryUrl, sqlQuery);
+ // It is okay to block here for the result because MSQ tasks return the task Id associated with it, which shouldn't
+ // consume a lot of time
+ StatusResponseHolder statusResponseHolder = responseHolderFuture.get();
+
+ // Check if the task has been accepted successfully
+ HttpResponseStatus httpResponseStatus = statusResponseHolder.getStatus();
+ if (!httpResponseStatus.equals(HttpResponseStatus.ACCEPTED)) {
+ throw new ISE(
+ "Unable to submit the task successfully. Received response status code [%d], and response content:\n[%s]",
+ httpResponseStatus,
+ statusResponseHolder.getContent()
+ );
+ }
+ String content = statusResponseHolder.getContent();
+ SqlTaskStatus sqlTaskStatus;
+ try {
+ sqlTaskStatus = jsonMapper.readValue(content, SqlTaskStatus.class);
+ }
+ catch (JsonProcessingException e) {
+ throw new ISE("Unable to parse the response");
+ }
+ if (sqlTaskStatus.getState().isFailure()) {
+ throw new ISE("Unable to start the task successfully.\nPossible exception: %s", sqlTaskStatus.getError());
+ }
+ return sqlTaskStatus.getTaskId();
+ }
+
+ /**
+ * Polls the overlord API every 1 second and waits for a submitted MSQ task to be completed. Alternatively, one can
+ * specify the maximum time to poll. The method returns the last fetched {@link TaskState} of the task
+ */
+ public TaskState pollTaskIdForCompletion(String taskId, long maxTimeoutSeconds)
+ {
+ if (maxTimeoutSeconds < 0) {
+ throw new IAE("Timeout cannot be negative");
+ } else if (maxTimeoutSeconds == 0) {
+ maxTimeoutSeconds = Long.MAX_VALUE;
+ }
+ long time = 0;
+ do {
+ TaskStatusPlus taskStatusPlus = overlordClient.getTaskStatus(taskId);
+ TaskState statusCode = taskStatusPlus.getStatusCode();
+ if (statusCode != null && statusCode.isComplete()) {
+ return taskStatusPlus.getStatusCode();
+ }
+ try {
+ Thread.sleep(1000);
+ }
+ catch (InterruptedException e) {
+ throw new ISE(e, "Interrupted while polling for task [%s] completion", taskId);
+ }
+ } while (time++ < maxTimeoutSeconds);
Review Comment:
This is a very weird and error-prone way of enforcing the timeout. The `RetryUtils` are probably already doing this in a cleaner way. Please try to reuse the approach used in the other ITs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] LakshSingla commented on a diff in pull request #12992: Add IT for MSQ task engine using the new IT framework
Posted by GitBox <gi...@apache.org>.
LakshSingla commented on code in PR #12992:
URL: https://github.com/apache/druid/pull/12992#discussion_r956914765
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/MsqQueryWithResults.java:
##########
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.utils;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+public class MsqQueryWithResults extends AbstractQueryWithResults<String>
Review Comment:
IMO this class is just to provide the query and the results. I didn't consider the case with custom context parameters, however, I think we can add these as parameters in the methods that utilize this class. I will make the necessary updates.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] LakshSingla commented on a diff in pull request #12992: Add IT for MSQ task engine using the new IT framework
Posted by GitBox <gi...@apache.org>.
LakshSingla commented on code in PR #12992:
URL: https://github.com/apache/druid/pull/12992#discussion_r958021496
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/MsqTestQueryHelper.java:
##########
@@ -0,0 +1,235 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.utils;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import org.apache.druid.indexer.TaskState;
+import org.apache.druid.indexer.TaskStatusPlus;
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.guava.Yielder;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.msq.indexing.report.MSQResultsReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReportPayload;
+import org.apache.druid.msq.sql.SqlTaskStatus;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.sql.http.SqlQuery;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.MsqOverlordResourceTestClient;
+import org.apache.druid.testing.clients.MsqTestClient;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
+import org.junit.Assert;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+/**
+ * Helper class to aid out ITs for MSQ
+ */
+public class MsqTestQueryHelper extends AbstractTestQueryHelper<MsqQueryWithResults>
+{
+
+ private final ObjectMapper jsonMapper;
+ private final IntegrationTestingConfig config;
+ private final MsqOverlordResourceTestClient overlordClient;
+ private final MsqTestClient msqClient;
+
+
+ @Inject
+ MsqTestQueryHelper(
+ final ObjectMapper jsonMapper,
+ final MsqTestClient queryClient,
+ final IntegrationTestingConfig config,
+ final MsqOverlordResourceTestClient overlordClient,
+ final MsqTestClient msqClient
+ )
+ {
+ super(jsonMapper, queryClient, config);
+ this.jsonMapper = jsonMapper;
+ this.config = config;
+ this.overlordClient = overlordClient;
+ this.msqClient = msqClient;
+ }
+
+ @Override
+ public String getQueryURL(String schemeAndHost)
+ {
+ return StringUtils.format("%s/druid/v2/sql/task", schemeAndHost);
+ }
+
+ /**
+ * Submits a task to the MSQ API with the given query string, and default headers and parameters
+ */
+ public String submitMsqTask(String sqlQueryString) throws ExecutionException, InterruptedException
+ {
+ return submitMsqTask(new SqlQuery(sqlQueryString, null, false, false, false, ImmutableMap.of(), null));
+ }
+
+ // Run the task, wait for it to complete, fetch the reports, verify the results,
+
+ /**
+ * Submits a {@link SqlQuery} to the MSQ API for execution. This method waits for the task to be accepted by the cluster
+ * and returns the task id associated with the submitted task
+ */
+ public String submitMsqTask(SqlQuery sqlQuery) throws ExecutionException, InterruptedException
+ {
+ String queryUrl = getQueryURL(config.getBrokerUrl());
+ Future<StatusResponseHolder> responseHolderFuture = msqClient.queryAsync(queryUrl, sqlQuery);
+ // It is okay to block here for the result because MSQ tasks return the task Id associated with it, which shouldn't
+ // consume a lot of time
+ StatusResponseHolder statusResponseHolder = responseHolderFuture.get();
+
+ // Check if the task has been accepted successfully
+ HttpResponseStatus httpResponseStatus = statusResponseHolder.getStatus();
+ if (!httpResponseStatus.equals(HttpResponseStatus.ACCEPTED)) {
+ throw new ISE(
+ "Unable to submit the task successfully. Received response status code [%d], and response content:\n[%s]",
+ httpResponseStatus,
+ statusResponseHolder.getContent()
+ );
+ }
+ String content = statusResponseHolder.getContent();
+ SqlTaskStatus sqlTaskStatus;
+ try {
+ sqlTaskStatus = jsonMapper.readValue(content, SqlTaskStatus.class);
+ }
+ catch (JsonProcessingException e) {
+ throw new ISE("Unable to parse the response");
+ }
+ if (sqlTaskStatus.getState().isFailure()) {
+ throw new ISE("Unable to start the task successfully.\nPossible exception: %s", sqlTaskStatus.getError());
+ }
+ return sqlTaskStatus.getTaskId();
+ }
+
+ /**
+ * Polls the overlord API every 1 second and waits for a submitted MSQ task to be completed. Alternatively, one can
+ * specify the maximum time to poll. The method returns the last fetched {@link TaskState} of the task
+ */
+ public TaskState pollTaskIdForCompletion(String taskId, long maxTimeoutSeconds)
Review Comment:
Rewritten the code using `RetryUtils` so that it is more error resilient. The only disadvantage that I can think off is the exponential backoff nature of the retries which I was trying to avoid.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] LakshSingla commented on a diff in pull request #12992: Add IT for MSQ task engine using the new IT framework
Posted by GitBox <gi...@apache.org>.
LakshSingla commented on code in PR #12992:
URL: https://github.com/apache/druid/pull/12992#discussion_r958011341
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/MsqQueryWithResults.java:
##########
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.utils;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+public class MsqQueryWithResults extends AbstractQueryWithResults<String>
Review Comment:
For custom context parameters, the task needs to be submitted with those in the `submitMsqTask` method. This POJO is used in `compareResults` method which merely fetches the reports for the submitted task and compares the result with the ones present in the report.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] paul-rogers commented on pull request #12992: Add IT for MSQ task engine using the new IT framework
Posted by GitBox <gi...@apache.org>.
paul-rogers commented on PR #12992:
URL: https://github.com/apache/druid/pull/12992#issuecomment-1249987215
An earlier comment mentioned that it would be handy to use the file format from the oft-discussed, but not yet committed planner test framework. To make this more concrete, I just opened a [PR #13106](https://github.com/apache/druid/pull/13106) with just the test file format, without the unit test mechanisms. This provides the option, in this PR, to use the proposed file format. The query file format is designed to validate the planer and so has many sections. Only a few are relevant to this PR. The thought is it might be simpler to reuse a subset of the test framework than it would be to reinvent the bits needed for an IT.
A typical test case might look like this:
```
=== case
An example insert test
=== sql
INSERT INTO myTable
SELECT ...
```
A `SELECT` query can include JSON-formatted results.
With this file format, a test would:
* Load the test file
* Loop over each test case
* Grab the SQL and execute it. If there is a need, you can also grab the query context.
* For an INSERT, just make sure the statement succeeds.
* For a SELECT, verify that the actual results match the expected results, and that the schema matches the expected schema.
The test framework PR includes an "actual results" class, but that might be overkill for the needs of an IT, since the IT will check only one or two expected results.
The mechanism does have code to verify results and display errors. When run in a UT, the framework produces an "actual" output file using the same format as the "expected" file. This makes it easy to use a diff tool to see errors and to capture new expected values. That won't work in an IT context, but we can devise something simpler, maybe just showing the actual results and the line at which the results differ. (The test code will provide an error with the line number.)
Perhaps there is a need to validate the reports. This can be done by adding a new section which would probably use the regex feature that turns a line into a regex if the line starts with "!". Also, "**" matches any number of lines:
```text
=== reports
{
"status": "SUCCESS",
! "runTimeMs": \d+,
**
...
```
Anyway, take a look. The code can be a bit daunting on first inspection, but the interface should be pretty simple. See the unit tests for examples.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] LakshSingla commented on a diff in pull request #12992: Add IT for MSQ task engine using the new IT framework
Posted by GitBox <gi...@apache.org>.
LakshSingla commented on code in PR #12992:
URL: https://github.com/apache/druid/pull/12992#discussion_r973811077
##########
integration-tests-ex/cases/src/test/resources/multi-stage-query/wikipedia_msq_select_query1.json:
##########
@@ -0,0 +1,31 @@
+[
Review Comment:
Updated!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] LakshSingla commented on a diff in pull request #12992: Add IT for MSQ task engine using the new IT framework
Posted by GitBox <gi...@apache.org>.
LakshSingla commented on code in PR #12992:
URL: https://github.com/apache/druid/pull/12992#discussion_r972847338
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/QueryResultVerifier.java:
##########
@@ -19,13 +19,19 @@
package org.apache.druid.testing.utils;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.logger.Logger;
+
import java.util.Iterator;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
public class QueryResultVerifier
{
- public static boolean compareResults(
+ private static final Logger LOG = new Logger(QueryResultVerifier.class);
+
+ public static Optional<String> compareResults(
Review Comment:
That was my initial approach, however, I realised that if success is true, then the error message would be null and if success is false then the error message would be some nonnull value, therefore having the success status seemed redundant to me. I can make the desired change while addressing the review comments.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] abhishekagarwal87 commented on pull request #12992: Add IT for MSQ task engine using the new IT framework
Posted by GitBox <gi...@apache.org>.
abhishekagarwal87 commented on PR #12992:
URL: https://github.com/apache/druid/pull/12992#issuecomment-1254845122
It will be worthwhile to pick any larger refactoring outside this PR. I am going to merge this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] LakshSingla commented on a diff in pull request #12992: Add IT for MSQ task engine using the new IT framework
Posted by GitBox <gi...@apache.org>.
LakshSingla commented on code in PR #12992:
URL: https://github.com/apache/druid/pull/12992#discussion_r973844958
##########
integration-tests/src/main/java/org/apache/druid/testing/clients/msq/MsqOverlordResourceTestClient.java:
##########
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.clients.msq;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.inject.Inject;
+import org.apache.druid.guice.annotations.Json;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.http.client.HttpClient;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.msq.guice.MSQIndexingModule;
+import org.apache.druid.msq.indexing.report.MSQTaskReport;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.OverlordResourceTestClient;
+import org.apache.druid.testing.guice.TestClient;
+import org.jboss.netty.handler.codec.http.HttpMethod;
+
+import java.util.Map;
+
+/**
+ * Overlord resource client for MSQ Tasks
+ */
+public class MsqOverlordResourceTestClient extends OverlordResourceTestClient
+{
+ ObjectMapper jsonMapper;
+
+ @Inject
+ MsqOverlordResourceTestClient(
+ @Json ObjectMapper jsonMapper,
+ @TestClient HttpClient httpClient,
+ IntegrationTestingConfig config
+ )
+ {
+ super(jsonMapper, httpClient, config);
+ this.jsonMapper = jsonMapper;
+ this.jsonMapper.registerModules(new MSQIndexingModule().getJacksonModules());
+ }
+
+ public Map<String, MSQTaskReport> getTaskReportForMsqTask(String taskId)
Review Comment:
This method doesn't rely on the structure of the MSQ's task report to fetch the results. The map is (from my understanding) how the reports API returns the report for a task.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] LakshSingla commented on a diff in pull request #12992: Add IT for MSQ task engine using the new IT framework
Posted by GitBox <gi...@apache.org>.
LakshSingla commented on code in PR #12992:
URL: https://github.com/apache/druid/pull/12992#discussion_r973824441
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/MsqTestQueryHelper.java:
##########
@@ -0,0 +1,246 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.utils;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import org.apache.druid.indexer.TaskState;
+import org.apache.druid.indexer.TaskStatusPlus;
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.RetryUtils;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.guava.Yielder;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.msq.indexing.report.MSQResultsReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReportPayload;
+import org.apache.druid.msq.sql.SqlTaskStatus;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.sql.http.SqlQuery;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.SqlResourceTestClient;
+import org.apache.druid.testing.clients.msq.MsqOverlordResourceTestClient;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+/**
+ * Helper class to aid out ITs for MSQ.
+ * This takes all the clients required to make the necessary calls to the client APIs for MSQ and performs the boilerplate
+ * functions for the tests
+ */
+public class MsqTestQueryHelper extends AbstractTestQueryHelper<MsqQueryWithResults>
+{
+
+ private final ObjectMapper jsonMapper;
+ private final IntegrationTestingConfig config;
+ private final MsqOverlordResourceTestClient overlordClient;
+ private final SqlResourceTestClient msqClient;
+
+
+ @Inject
+ MsqTestQueryHelper(
+ final ObjectMapper jsonMapper,
+ final SqlResourceTestClient queryClient,
+ final IntegrationTestingConfig config,
+ final MsqOverlordResourceTestClient overlordClient,
+ final SqlResourceTestClient msqClient
+ )
+ {
+ super(jsonMapper, queryClient, config);
+ this.jsonMapper = jsonMapper;
+ this.config = config;
+ this.overlordClient = overlordClient;
+ this.msqClient = msqClient;
+ }
+
+ @Override
+ public String getQueryURL(String schemeAndHost)
+ {
+ return StringUtils.format("%s/druid/v2/sql/task", schemeAndHost);
+ }
+
+ /**
+ * Submits a task to the MSQ API with the given query string, and default headers and parameters
+ */
+ public String submitMsqTask(String sqlQueryString) throws ExecutionException, InterruptedException
+ {
+ return submitMsqTask(new SqlQuery(sqlQueryString, null, false, false, false, ImmutableMap.of(), null));
+ }
+
+ // Run the task, wait for it to complete, fetch the reports, verify the results,
+
+ /**
+ * Submits a {@link SqlQuery} to the MSQ API for execution. This method waits for the task to be accepted by the cluster
+ * and returns the task id associated with the submitted task
+ */
+ public String submitMsqTask(SqlQuery sqlQuery) throws ExecutionException, InterruptedException
+ {
+ String queryUrl = getQueryURL(config.getBrokerUrl());
+ Future<StatusResponseHolder> responseHolderFuture = msqClient.queryAsync(queryUrl, sqlQuery);
+ // It is okay to block here for the result because MSQ tasks return the task Id associated with it, which shouldn't
+ // consume a lot of time
+ StatusResponseHolder statusResponseHolder = responseHolderFuture.get();
Review Comment:
Added a failsafe timeout of 5 minutes so in case we are unable to fetch the status id in time.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] cryptoe commented on a diff in pull request #12992: Add IT for MSQ task engine using the new IT framework
Posted by GitBox <gi...@apache.org>.
cryptoe commented on code in PR #12992:
URL: https://github.com/apache/druid/pull/12992#discussion_r956891634
##########
integration-tests/src/main/java/org/apache/druid/testing/clients/MsqOverlordResourceTestClient.java:
##########
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.clients;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableList;
+import com.google.inject.Inject;
+import org.apache.druid.guice.annotations.Json;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.http.client.HttpClient;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.msq.guice.MSQIndexingModule;
+import org.apache.druid.msq.indexing.report.MSQTaskReport;
+import org.apache.druid.segment.virtual.ExpressionVirtualColumn;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.guice.TestClient;
+import org.jboss.netty.handler.codec.http.HttpMethod;
+
+import java.util.Map;
+
+/**
+ * Overlord resource client for MSQ Tasks
+ */
+public class MsqOverlordResourceTestClient extends OverlordResourceTestClient
+{
+ ObjectMapper jsonMapper;
+
+ @Inject
+ MsqOverlordResourceTestClient(
+ @Json ObjectMapper jsonMapper,
+ @TestClient HttpClient httpClient,
+ IntegrationTestingConfig config
+ )
+ {
+ super(jsonMapper, httpClient, config);
+ this.jsonMapper = jsonMapper;
+ }
+
+ public Map<String, MSQTaskReport> getTaskReportForMsqTask(String taskId)
+ {
+ try {
+ StatusResponseHolder response = makeRequest(
+ HttpMethod.GET,
+ StringUtils.format(
+ "%s%s",
+ getIndexerURL(),
+ StringUtils.format("task/%s/reports", StringUtils.urlEncode(taskId))
+ )
+ );
+ jsonMapper.registerModules(new MSQIndexingModule().getJacksonModules());
Review Comment:
IMHO As this is called per method call, we should remove this registration from here and add it either to
1. Where we are adding the modules
2. Or the constructor.
##########
integration-tests/pom.xml:
##########
@@ -239,6 +239,12 @@
<artifactId>simple-client-sslcontext</artifactId>
<version>${project.parent.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.druid.extensions</groupId>
+ <artifactId>druid-multi-stage-query</artifactId>
+ <version>${project.parent.version}</version>
+<!-- <scope>test</scope>-->
Review Comment:
Nit: I think we should remove this comment out code.
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/MsqQueryWithResults.java:
##########
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.utils;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+public class MsqQueryWithResults extends AbstractQueryWithResults<String>
Review Comment:
How do we pass custom context parameters?
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/MsqTestQueryHelper.java:
##########
@@ -0,0 +1,235 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.utils;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import org.apache.druid.indexer.TaskState;
+import org.apache.druid.indexer.TaskStatusPlus;
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.guava.Yielder;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.msq.indexing.report.MSQResultsReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReportPayload;
+import org.apache.druid.msq.sql.SqlTaskStatus;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.sql.http.SqlQuery;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.MsqOverlordResourceTestClient;
+import org.apache.druid.testing.clients.MsqTestClient;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
+import org.junit.Assert;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+/**
+ * Helper class to aid out ITs for MSQ
+ */
+public class MsqTestQueryHelper extends AbstractTestQueryHelper<MsqQueryWithResults>
+{
+
+ private final ObjectMapper jsonMapper;
+ private final IntegrationTestingConfig config;
+ private final MsqOverlordResourceTestClient overlordClient;
+ private final MsqTestClient msqClient;
+
+
+ @Inject
+ MsqTestQueryHelper(
+ final ObjectMapper jsonMapper,
+ final MsqTestClient queryClient,
+ final IntegrationTestingConfig config,
+ final MsqOverlordResourceTestClient overlordClient,
+ final MsqTestClient msqClient
+ )
+ {
+ super(jsonMapper, queryClient, config);
+ this.jsonMapper = jsonMapper;
+ this.config = config;
+ this.overlordClient = overlordClient;
+ this.msqClient = msqClient;
+ }
+
+ @Override
+ public String getQueryURL(String schemeAndHost)
+ {
+ return StringUtils.format("%s/druid/v2/sql/task", schemeAndHost);
+ }
+
+ /**
+ * Submits a task to the MSQ API with the given query string, and default headers and parameters
+ */
+ public String submitMsqTask(String sqlQueryString) throws ExecutionException, InterruptedException
+ {
+ return submitMsqTask(new SqlQuery(sqlQueryString, null, false, false, false, ImmutableMap.of(), null));
+ }
+
+ // Run the task, wait for it to complete, fetch the reports, verify the results,
+
+ /**
+ * Submits a {@link SqlQuery} to the MSQ API for execution. This method waits for the task to be accepted by the cluster
+ * and returns the task id associated with the submitted task
+ */
+ public String submitMsqTask(SqlQuery sqlQuery) throws ExecutionException, InterruptedException
+ {
+ String queryUrl = getQueryURL(config.getBrokerUrl());
+ Future<StatusResponseHolder> responseHolderFuture = msqClient.queryAsync(queryUrl, sqlQuery);
+ // It is okay to block here for the result because MSQ tasks return the task Id associated with it, which shouldn't
+ // consume a lot of time
+ StatusResponseHolder statusResponseHolder = responseHolderFuture.get();
+
+ // Check if the task has been accepted successfully
+ HttpResponseStatus httpResponseStatus = statusResponseHolder.getStatus();
+ if (!httpResponseStatus.equals(HttpResponseStatus.ACCEPTED)) {
+ throw new ISE(
+ "Unable to submit the task successfully. Received response status code [%d], and response content:\n[%s]",
+ httpResponseStatus,
+ statusResponseHolder.getContent()
+ );
+ }
+ String content = statusResponseHolder.getContent();
+ SqlTaskStatus sqlTaskStatus;
+ try {
+ sqlTaskStatus = jsonMapper.readValue(content, SqlTaskStatus.class);
+ }
+ catch (JsonProcessingException e) {
+ throw new ISE("Unable to parse the response");
+ }
+ if (sqlTaskStatus.getState().isFailure()) {
+ throw new ISE("Unable to start the task successfully.\nPossible exception: %s", sqlTaskStatus.getError());
+ }
+ return sqlTaskStatus.getTaskId();
+ }
+
+ /**
+ * Polls the overlord API every 1 second and waits for a submitted MSQ task to be completed. Alternatively, one can
+ * specify the maximum time to poll. The method returns the last fetched {@link TaskState} of the task
+ */
+ public TaskState pollTaskIdForCompletion(String taskId, long maxTimeoutSeconds)
Review Comment:
We would never want the IT's to hang so I think we should dis allow "maxTimeouts"
Also what would happen if the taskId is not complete? I feel that the caller of this method would assume the taskID to always complete in the timeout. Should we throw an exception if the task is not completed?
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/MsqTestQueryHelper.java:
##########
@@ -0,0 +1,235 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.utils;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import org.apache.druid.indexer.TaskState;
+import org.apache.druid.indexer.TaskStatusPlus;
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.guava.Yielder;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.msq.indexing.report.MSQResultsReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReportPayload;
+import org.apache.druid.msq.sql.SqlTaskStatus;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.sql.http.SqlQuery;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.MsqOverlordResourceTestClient;
+import org.apache.druid.testing.clients.MsqTestClient;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
+import org.junit.Assert;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+/**
+ * Helper class to aid out ITs for MSQ
+ */
+public class MsqTestQueryHelper extends AbstractTestQueryHelper<MsqQueryWithResults>
+{
+
+ private final ObjectMapper jsonMapper;
+ private final IntegrationTestingConfig config;
+ private final MsqOverlordResourceTestClient overlordClient;
+ private final MsqTestClient msqClient;
+
+
+ @Inject
+ MsqTestQueryHelper(
+ final ObjectMapper jsonMapper,
+ final MsqTestClient queryClient,
+ final IntegrationTestingConfig config,
+ final MsqOverlordResourceTestClient overlordClient,
+ final MsqTestClient msqClient
+ )
+ {
+ super(jsonMapper, queryClient, config);
+ this.jsonMapper = jsonMapper;
+ this.config = config;
+ this.overlordClient = overlordClient;
+ this.msqClient = msqClient;
+ }
+
+ @Override
+ public String getQueryURL(String schemeAndHost)
+ {
+ return StringUtils.format("%s/druid/v2/sql/task", schemeAndHost);
+ }
+
+ /**
+ * Submits a task to the MSQ API with the given query string, and default headers and parameters
+ */
+ public String submitMsqTask(String sqlQueryString) throws ExecutionException, InterruptedException
+ {
+ return submitMsqTask(new SqlQuery(sqlQueryString, null, false, false, false, ImmutableMap.of(), null));
+ }
+
+ // Run the task, wait for it to complete, fetch the reports, verify the results,
+
+ /**
+ * Submits a {@link SqlQuery} to the MSQ API for execution. This method waits for the task to be accepted by the cluster
+ * and returns the task id associated with the submitted task
+ */
+ public String submitMsqTask(SqlQuery sqlQuery) throws ExecutionException, InterruptedException
+ {
+ String queryUrl = getQueryURL(config.getBrokerUrl());
+ Future<StatusResponseHolder> responseHolderFuture = msqClient.queryAsync(queryUrl, sqlQuery);
+ // It is okay to block here for the result because MSQ tasks return the task Id associated with it, which shouldn't
+ // consume a lot of time
+ StatusResponseHolder statusResponseHolder = responseHolderFuture.get();
+
+ // Check if the task has been accepted successfully
+ HttpResponseStatus httpResponseStatus = statusResponseHolder.getStatus();
+ if (!httpResponseStatus.equals(HttpResponseStatus.ACCEPTED)) {
+ throw new ISE(
+ "Unable to submit the task successfully. Received response status code [%d], and response content:\n[%s]",
+ httpResponseStatus,
+ statusResponseHolder.getContent()
+ );
+ }
+ String content = statusResponseHolder.getContent();
+ SqlTaskStatus sqlTaskStatus;
+ try {
+ sqlTaskStatus = jsonMapper.readValue(content, SqlTaskStatus.class);
+ }
+ catch (JsonProcessingException e) {
+ throw new ISE("Unable to parse the response");
+ }
+ if (sqlTaskStatus.getState().isFailure()) {
+ throw new ISE("Unable to start the task successfully.\nPossible exception: %s", sqlTaskStatus.getError());
+ }
+ return sqlTaskStatus.getTaskId();
+ }
+
+ /**
+ * Polls the overlord API every 1 second and waits for a submitted MSQ task to be completed. Alternatively, one can
+ * specify the maximum time to poll. The method returns the last fetched {@link TaskState} of the task
+ */
+ public TaskState pollTaskIdForCompletion(String taskId, long maxTimeoutSeconds)
Review Comment:
This piece of code would be already present in some indexing IT no ?
##########
integration-tests/src/main/java/org/apache/druid/testing/clients/MsqTestClient.java:
##########
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.clients;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.inject.Inject;
+import org.apache.druid.java.util.http.client.HttpClient;
+import org.apache.druid.sql.http.SqlQuery;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.guice.TestClient;
+
+import javax.ws.rs.core.MediaType;
+
+public class MsqTestClient extends AbstractQueryResourceTestClient<SqlQuery>
Review Comment:
Can't we use `QueryResourceTestClient` ?
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/MsqTestQueryHelper.java:
##########
@@ -0,0 +1,235 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.testing.utils;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import org.apache.druid.indexer.TaskState;
+import org.apache.druid.indexer.TaskStatusPlus;
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.guava.Yielder;
+import org.apache.druid.java.util.http.client.response.StatusResponseHolder;
+import org.apache.druid.msq.indexing.report.MSQResultsReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReport;
+import org.apache.druid.msq.indexing.report.MSQTaskReportPayload;
+import org.apache.druid.msq.sql.SqlTaskStatus;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.sql.http.SqlQuery;
+import org.apache.druid.testing.IntegrationTestingConfig;
+import org.apache.druid.testing.clients.MsqOverlordResourceTestClient;
+import org.apache.druid.testing.clients.MsqTestClient;
+import org.jboss.netty.handler.codec.http.HttpResponseStatus;
+import org.junit.Assert;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+/**
+ * Helper class to aid out ITs for MSQ
+ */
+public class MsqTestQueryHelper extends AbstractTestQueryHelper<MsqQueryWithResults>
+{
+
+ private final ObjectMapper jsonMapper;
+ private final IntegrationTestingConfig config;
+ private final MsqOverlordResourceTestClient overlordClient;
+ private final MsqTestClient msqClient;
+
+
+ @Inject
+ MsqTestQueryHelper(
+ final ObjectMapper jsonMapper,
+ final MsqTestClient queryClient,
+ final IntegrationTestingConfig config,
+ final MsqOverlordResourceTestClient overlordClient,
+ final MsqTestClient msqClient
+ )
+ {
+ super(jsonMapper, queryClient, config);
+ this.jsonMapper = jsonMapper;
+ this.config = config;
+ this.overlordClient = overlordClient;
+ this.msqClient = msqClient;
+ }
+
+ @Override
+ public String getQueryURL(String schemeAndHost)
+ {
+ return StringUtils.format("%s/druid/v2/sql/task", schemeAndHost);
+ }
+
+ /**
+ * Submits a task to the MSQ API with the given query string, and default headers and parameters
+ */
+ public String submitMsqTask(String sqlQueryString) throws ExecutionException, InterruptedException
+ {
+ return submitMsqTask(new SqlQuery(sqlQueryString, null, false, false, false, ImmutableMap.of(), null));
+ }
+
+ // Run the task, wait for it to complete, fetch the reports, verify the results,
+
+ /**
+ * Submits a {@link SqlQuery} to the MSQ API for execution. This method waits for the task to be accepted by the cluster
+ * and returns the task id associated with the submitted task
+ */
+ public String submitMsqTask(SqlQuery sqlQuery) throws ExecutionException, InterruptedException
+ {
+ String queryUrl = getQueryURL(config.getBrokerUrl());
+ Future<StatusResponseHolder> responseHolderFuture = msqClient.queryAsync(queryUrl, sqlQuery);
+ // It is okay to block here for the result because MSQ tasks return the task Id associated with it, which shouldn't
+ // consume a lot of time
+ StatusResponseHolder statusResponseHolder = responseHolderFuture.get();
+
+ // Check if the task has been accepted successfully
+ HttpResponseStatus httpResponseStatus = statusResponseHolder.getStatus();
+ if (!httpResponseStatus.equals(HttpResponseStatus.ACCEPTED)) {
+ throw new ISE(
+ "Unable to submit the task successfully. Received response status code [%d], and response content:\n[%s]",
+ httpResponseStatus,
+ statusResponseHolder.getContent()
+ );
+ }
+ String content = statusResponseHolder.getContent();
+ SqlTaskStatus sqlTaskStatus;
+ try {
+ sqlTaskStatus = jsonMapper.readValue(content, SqlTaskStatus.class);
+ }
+ catch (JsonProcessingException e) {
+ throw new ISE("Unable to parse the response");
+ }
+ if (sqlTaskStatus.getState().isFailure()) {
+ throw new ISE("Unable to start the task successfully.\nPossible exception: %s", sqlTaskStatus.getError());
+ }
+ return sqlTaskStatus.getTaskId();
+ }
+
+ /**
+ * Polls the overlord API every 1 second and waits for a submitted MSQ task to be completed. Alternatively, one can
+ * specify the maximum time to poll. The method returns the last fetched {@link TaskState} of the task
+ */
+ public TaskState pollTaskIdForCompletion(String taskId, long maxTimeoutSeconds)
Review Comment:
If we decide to keep this method, a line mentioning the various timeout cases in the javadocs would be useful.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org