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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
