diff --git a/druid-handler/pom.xml b/druid-handler/pom.xml index dcc5ebd8a481..8f94abea49e7 100644 --- a/druid-handler/pom.xml +++ b/druid-handler/pom.xml @@ -25,7 +25,7 @@ Hive Druid Handler .. - 16.0.1 + ${guava.version} @@ -34,6 +34,14 @@ com.fasterxml.jackson.dataformat jackson-dataformat-smile + + com.fasterxml.jackson.core + jackson-annotations + + + com.fasterxml.jackson.core + jackson-core + com.fasterxml.jackson.core jackson-databind @@ -77,13 +85,16 @@ ${druid.guava.version} - joda-time - joda-time + org.asynchttpclient + async-http-client - io.netty - netty - ${netty3.version} + org.apache.httpcomponents + httpclient + + + joda-time + joda-time org.apache.druid @@ -114,6 +125,10 @@ io.netty * + + org.asynchttpclient + async-http-client + org.glassfish javax.el @@ -130,6 +145,26 @@ com.ibm.icu icu4j + + com.fasterxml.jackson.core + jackson-core + + + com.fasterxml.jackson.core + jackson-annotations + + + com.fasterxml.jackson.core + jackson-databind + + + org.asynchttpclient + async-http-client + + + io.netty + netty + org.codehaus.plexus plexus-utils @@ -332,10 +367,10 @@ org.apache.maven.plugins maven-shade-plugin - - + + package @@ -357,10 +392,6 @@ org.apache.calcite org.apache.hive.druid.org.apache.calcite - - org.jboss.netty - org.apache.hive.druid.org.jboss.netty - com.fasterxml.jackson org.apache.hive.druid.com.fasterxml.jackson @@ -393,6 +424,10 @@ org.asynchttpclient:* org.antlr:* + + + io.netty:netty + diff --git a/druid-handler/src/java/org/apache/hadoop/hive/druid/DruidKafkaUtils.java b/druid-handler/src/java/org/apache/hadoop/hive/druid/DruidKafkaUtils.java index fb6ce308fbc7..e6024c3c1aee 100644 --- a/druid-handler/src/java/org/apache/hadoop/hive/druid/DruidKafkaUtils.java +++ b/druid-handler/src/java/org/apache/hadoop/hive/druid/DruidKafkaUtils.java @@ -28,13 +28,12 @@ import org.apache.druid.data.input.impl.JSONParseSpec; import org.apache.druid.data.input.impl.StringInputRowParser; import org.apache.druid.data.input.impl.TimestampSpec; -import org.apache.druid.java.util.http.client.Request; -import org.apache.druid.java.util.http.client.response.StringFullResponseHandler; -import org.apache.druid.java.util.http.client.response.StringFullResponseHolder; import org.apache.druid.segment.IndexSpec; import org.apache.druid.segment.indexing.DataSchema; import org.apache.druid.segment.writeout.TmpFileSegmentWriteOutMediumFactory; import org.apache.hadoop.hive.druid.conf.DruidConstants; +import org.apache.hadoop.hive.druid.http.HiveDruidHttpRequest; +import org.apache.hadoop.hive.druid.http.HiveDruidHttpResponse; import org.apache.hadoop.hive.druid.json.AvroParseSpec; import org.apache.hadoop.hive.druid.json.AvroStreamInputRowParser; import org.apache.hadoop.hive.druid.json.InlineSchemaAvroBytesDecoder; @@ -43,8 +42,6 @@ import org.apache.hadoop.hive.druid.json.KafkaSupervisorTuningConfig; import org.apache.hadoop.hive.metastore.api.Table; import org.apache.hadoop.hive.ql.session.SessionState; -import org.jboss.netty.handler.codec.http.HttpMethod; -import org.jboss.netty.handler.codec.http.HttpResponseStatus; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -159,22 +156,21 @@ static void updateKafkaIngestionSpec(String overlordAddress, KafkaSupervisorSpec String task = JSON_MAPPER.writeValueAsString(spec); CONSOLE.printInfo("submitting kafka Spec {}", task); LOG.info("submitting kafka Supervisor Spec {}", task); - StringFullResponseHolder response = DruidStorageHandlerUtils - .getResponseFromCurrentLeader(DruidStorageHandler.getHttpClient(), new Request(HttpMethod.POST, - new URL(String.format("http://%s/druid/indexer/v1/supervisor", overlordAddress))) - .setContent("application/json", JSON_MAPPER.writeValueAsBytes(spec)), - new StringFullResponseHandler(Charset.forName("UTF-8"))); - if (response.getStatus().equals(HttpResponseStatus.OK)) { - String - msg = - String.format("Kafka Supervisor for [%s] Submitted Successfully to druid.", - spec.getDataSchema().getDataSource()); + HiveDruidHttpResponse response = DruidStorageHandlerUtils.getResponseFromCurrentLeader( + DruidStorageHandler.getHttpClient(), + new HiveDruidHttpRequest("POST", + new URL(String.format("http://%s/druid/indexer/v1/supervisor", overlordAddress))) + .setContent(JSON_MAPPER.writeValueAsBytes(spec)) + .setHeader("Content-Type", "application/json")); + if (response.getStatusCode() == HiveDruidHttpResponse.SC_OK) { + String msg = String.format("Kafka Supervisor for [%s] Submitted Successfully to druid.", + spec.getDataSchema().getDataSource()); LOG.info(msg); CONSOLE.printInfo(msg); } else { - throw new IOException(String.format("Unable to update Kafka Ingestion for Druid status [%d] full response [%s]", - response.getStatus().getCode(), - response.getContent())); + throw new IOException(String.format( + "Unable to update Kafka Ingestion for Druid status [%d] full response [%s]", + response.getStatusCode(), response.getContent())); } } catch (Exception e) { throw new RuntimeException(e); @@ -191,7 +187,7 @@ static InputRowParser getInputRowParser(Table table, TimestampSpec timestampSpec // Default case JSON if ((parseSpecFormat == null) || "json".equalsIgnoreCase(parseSpecFormat)) { - return new StringInputRowParser(new JSONParseSpec(timestampSpec, dimensionsSpec, null, null), "UTF-8"); + return new StringInputRowParser(new JSONParseSpec(timestampSpec, dimensionsSpec), "UTF-8"); } else if ("csv".equalsIgnoreCase(parseSpecFormat)) { return new StringInputRowParser(new CSVParseSpec(timestampSpec, dimensionsSpec, diff --git a/druid-handler/src/java/org/apache/hadoop/hive/druid/DruidStorageHandler.java b/druid-handler/src/java/org/apache/hadoop/hive/druid/DruidStorageHandler.java index 656fa40c03fa..856b58eae08a 100644 --- a/druid-handler/src/java/org/apache/hadoop/hive/druid/DruidStorageHandler.java +++ b/druid-handler/src/java/org/apache/hadoop/hive/druid/DruidStorageHandler.java @@ -34,20 +34,17 @@ import org.apache.druid.data.input.impl.TimestampSpec; import org.apache.druid.java.util.common.Pair; import org.apache.druid.java.util.common.RetryUtils; -import org.apache.druid.java.util.common.lifecycle.Lifecycle; -import org.apache.druid.java.util.http.client.HttpClient; -import org.apache.druid.java.util.http.client.HttpClientConfig; -import org.apache.druid.java.util.http.client.HttpClientInit; -import org.apache.druid.java.util.http.client.Request; -import org.apache.druid.java.util.http.client.response.StringFullResponseHandler; -import org.apache.druid.java.util.http.client.response.StringFullResponseHolder; +import org.apache.hadoop.hive.druid.http.HiveDruidHttpClient; +import org.apache.hadoop.hive.druid.http.HiveDruidHttpRequest; +import org.apache.hadoop.hive.druid.http.HiveDruidHttpResponse; import org.apache.druid.metadata.MetadataStorageConnectorConfig; import org.apache.druid.metadata.MetadataStorageTablesConfig; import org.apache.druid.metadata.SQLMetadataConnector; import org.apache.druid.metadata.storage.derby.DerbyConnector; import org.apache.druid.metadata.storage.derby.DerbyMetadataStorage; import org.apache.druid.metadata.storage.mysql.MySQLConnector; -import org.apache.druid.metadata.storage.mysql.MySQLConnectorConfig; +import org.apache.druid.metadata.storage.mysql.MySQLConnectorDriverConfig; +import org.apache.druid.metadata.storage.mysql.MySQLConnectorSslConfig; import org.apache.druid.metadata.storage.postgresql.PostgreSQLConnector; import org.apache.druid.metadata.storage.postgresql.PostgreSQLConnectorConfig; import org.apache.druid.metadata.storage.postgresql.PostgreSQLTablesConfig; @@ -75,7 +72,6 @@ import org.apache.hadoop.hive.druid.io.DruidRecordWriter; import org.apache.hadoop.hive.druid.json.KafkaSupervisorReport; import org.apache.hadoop.hive.druid.json.KafkaSupervisorSpec; -import org.apache.hadoop.hive.druid.security.KerberosHttpClient; import org.apache.hadoop.hive.druid.serde.DruidSerDe; import org.apache.hadoop.hive.metastore.DefaultHiveMetaHook; import org.apache.hadoop.hive.metastore.HiveMetaHook; @@ -105,8 +101,6 @@ import org.apache.hadoop.mapred.OutputFormat; import org.apache.hadoop.security.UserGroupInformation; import org.apache.hive.common.util.ShutdownHookManager; -import org.jboss.netty.handler.codec.http.HttpMethod; -import org.jboss.netty.handler.codec.http.HttpResponseStatus; import org.joda.time.DateTime; import org.joda.time.Period; import org.skife.jdbi.v2.exceptions.CallbackFailedException; @@ -119,7 +113,6 @@ import java.net.URL; import java.net.URI; import java.net.URISyntaxException; -import java.nio.charset.Charset; import java.util.ArrayList; import java.util.Collection; import java.util.Collections; @@ -127,7 +120,6 @@ import java.util.Map; import java.util.Properties; import java.util.Set; -import java.util.concurrent.ExecutionException; import java.util.stream.Collectors; import static org.apache.hadoop.hive.druid.DruidStorageHandlerUtils.JSON_MAPPER; @@ -146,7 +138,7 @@ private static final String INTERMEDIATE_SEGMENT_DIR_NAME = "intermediateSegmentDir"; - private static final HttpClient HTTP_CLIENT; + private static final HiveDruidHttpClient HTTP_CLIENT; private static final List ALLOWED_ALTER_TYPES = ImmutableList.of("ADDPROPS", "DROPPROPS", "ADDCOLS"); @@ -157,14 +149,14 @@ private static final String DRUID_HOST_NAME = "druid.zk.service.host"; static { - final Lifecycle lifecycle = new Lifecycle(); - try { - lifecycle.start(); - } catch (Exception e) { - LOG.error("Issues with lifecycle start", e); - } - HTTP_CLIENT = makeHttpClient(lifecycle); - ShutdownHookManager.addShutdownHook(lifecycle::stop); + HTTP_CLIENT = makeHttpClient(); + ShutdownHookManager.addShutdownHook(() -> { + try { + HTTP_CLIENT.close(); + } catch (IOException e) { + LOG.warn("Failed to close HiveDruidHttpClient", e); + } + }); } private SQLMetadataConnector connector; @@ -319,7 +311,7 @@ private void updateKafkaIngestion(Table table) { + columnNames); } - DimensionsSpec dimensionsSpec = new DimensionsSpec(dimensionsAndAggregates.lhs, null, null); + DimensionsSpec dimensionsSpec = new DimensionsSpec(dimensionsAndAggregates.lhs); String timestampFormat = DruidStorageHandlerUtils .getTableProperty(table, DruidConstants.DRUID_TIMESTAMP_FORMAT); String timestampColumnName = DruidStorageHandlerUtils @@ -388,19 +380,16 @@ private void updateKafkaIngestion(Table table) { private void resetKafkaIngestion(String overlordAddress, String dataSourceName) { try { - StringFullResponseHolder - response = - RetryUtils.retry(() -> DruidStorageHandlerUtils.getResponseFromCurrentLeader(getHttpClient(), - new Request(HttpMethod.POST, new URL( - String.format("http://%s/druid/indexer/v1/supervisor/%s/reset", overlordAddress, dataSourceName))), - new StringFullResponseHandler(Charset.forName("UTF-8"))), input -> input instanceof IOException, - getMaxRetryCount()); - if (response.getStatus().equals(HttpResponseStatus.OK)) { + HiveDruidHttpResponse response = RetryUtils.retry(() -> + DruidStorageHandlerUtils.getResponseFromCurrentLeader(getHttpClient(), + new HiveDruidHttpRequest("POST", new URL( + String.format("http://%s/druid/indexer/v1/supervisor/%s/reset", overlordAddress, dataSourceName)))), + input -> input instanceof IOException, getMaxRetryCount()); + if (response.getStatusCode() == HiveDruidHttpResponse.SC_OK) { CONSOLE.printInfo("Druid Kafka Ingestion Reset successful."); } else { throw new IOException(String.format("Unable to reset Kafka Ingestion Druid status [%d] full response [%s]", - response.getStatus().getCode(), - response.getContent())); + response.getStatusCode(), response.getContent())); } } catch (Exception e) { throw new RuntimeException(e); @@ -409,19 +398,16 @@ private void resetKafkaIngestion(String overlordAddress, String dataSourceName) private void stopKafkaIngestion(String overlordAddress, String dataSourceName) { try { - StringFullResponseHolder - response = - RetryUtils.retry(() -> DruidStorageHandlerUtils.getResponseFromCurrentLeader(getHttpClient(), - new Request(HttpMethod.POST, new URL( - String.format("http://%s/druid/indexer/v1/supervisor/%s/shutdown", overlordAddress, dataSourceName))), - new StringFullResponseHandler(Charset.forName("UTF-8"))), input -> input instanceof IOException, - getMaxRetryCount()); - if (response.getStatus().equals(HttpResponseStatus.OK)) { + HiveDruidHttpResponse response = RetryUtils.retry(() -> + DruidStorageHandlerUtils.getResponseFromCurrentLeader(getHttpClient(), + new HiveDruidHttpRequest("POST", new URL( + String.format("http://%s/druid/indexer/v1/supervisor/%s/shutdown", overlordAddress, dataSourceName)))), + input -> input instanceof IOException, getMaxRetryCount()); + if (response.getStatusCode() == HiveDruidHttpResponse.SC_OK) { CONSOLE.printInfo("Druid Kafka Ingestion shutdown successful."); } else { throw new IOException(String.format("Unable to stop Kafka Ingestion Druid status [%d] full response [%s]", - response.getStatus().getCode(), - response.getContent())); + response.getStatusCode(), response.getContent())); } } catch (Exception e) { throw new RuntimeException(e); @@ -440,25 +426,22 @@ private KafkaSupervisorSpec fetchKafkaIngestionSpec(Table table) { Preconditions.checkNotNull(DruidStorageHandlerUtils.getTableProperty(table, Constants.DRUID_DATA_SOURCE), "Druid Datasource name is null"); try { - StringFullResponseHolder - response = - RetryUtils.retry(() -> DruidStorageHandlerUtils.getResponseFromCurrentLeader(getHttpClient(), - new Request(HttpMethod.GET, - new URL(String.format("http://%s/druid/indexer/v1/supervisor/%s", overlordAddress, dataSourceName))), - new StringFullResponseHandler(Charset.forName("UTF-8"))), input -> input instanceof IOException, - getMaxRetryCount()); - if (response.getStatus().equals(HttpResponseStatus.OK)) { + HiveDruidHttpResponse response = RetryUtils.retry(() -> + DruidStorageHandlerUtils.getResponseFromCurrentLeader(getHttpClient(), + new HiveDruidHttpRequest("GET", + new URL(String.format("http://%s/druid/indexer/v1/supervisor/%s", overlordAddress, dataSourceName)))), + input -> input instanceof IOException, getMaxRetryCount()); + if (response.getStatusCode() == HiveDruidHttpResponse.SC_OK) { return JSON_MAPPER.readValue(response.getContent(), KafkaSupervisorSpec.class); // Druid Returns 400 Bad Request when not found. - } else if (response.getStatus().equals(HttpResponseStatus.NOT_FOUND) || response.getStatus() - .equals(HttpResponseStatus.BAD_REQUEST)) { + } else if (response.getStatusCode() == HiveDruidHttpResponse.SC_NOT_FOUND + || response.getStatusCode() == HiveDruidHttpResponse.SC_BAD_REQUEST) { LOG.debug("No Kafka Supervisor found for datasource[%s]", dataSourceName); return null; } else { throw new IOException(String.format( "Unable to fetch Kafka Ingestion Spec from Druid status [%d] full response [%s]", - response.getStatus().getCode(), - response.getContent())); + response.getStatusCode(), response.getContent())); } } catch (Exception e) { throw new RuntimeException("Exception while fetching kafka ingestion spec from druid", e); @@ -481,24 +464,21 @@ private KafkaSupervisorSpec fetchKafkaIngestionSpec(Table table) { Preconditions.checkNotNull(DruidStorageHandlerUtils.getTableProperty(table, Constants.DRUID_DATA_SOURCE), "Druid Datasource name is null"); try { - StringFullResponseHolder - response = - RetryUtils.retry(() -> DruidStorageHandlerUtils.getResponseFromCurrentLeader(getHttpClient(), - new Request(HttpMethod.GET, new URL( - String.format("http://%s/druid/indexer/v1/supervisor/%s/status", overlordAddress, dataSourceName))), - new StringFullResponseHandler(Charset.forName("UTF-8"))), input -> input instanceof IOException, - getMaxRetryCount()); - if (response.getStatus().equals(HttpResponseStatus.OK)) { + HiveDruidHttpResponse response = RetryUtils.retry(() -> + DruidStorageHandlerUtils.getResponseFromCurrentLeader(getHttpClient(), + new HiveDruidHttpRequest("GET", new URL( + String.format("http://%s/druid/indexer/v1/supervisor/%s/status", overlordAddress, dataSourceName)))), + input -> input instanceof IOException, getMaxRetryCount()); + if (response.getStatusCode() == HiveDruidHttpResponse.SC_OK) { return DruidStorageHandlerUtils.JSON_MAPPER.readValue(response.getContent(), KafkaSupervisorReport.class); // Druid Returns 400 Bad Request when not found. - } else if (response.getStatus().equals(HttpResponseStatus.NOT_FOUND) || response.getStatus() - .equals(HttpResponseStatus.BAD_REQUEST)) { + } else if (response.getStatusCode() == HiveDruidHttpResponse.SC_NOT_FOUND + || response.getStatusCode() == HiveDruidHttpResponse.SC_BAD_REQUEST) { LOG.info("No Kafka Supervisor found for datasource[%s]", dataSourceName); return null; } else { - LOG.error("Unable to fetch Kafka Supervisor status [%d] full response [%s]", - response.getStatus().getCode(), - response.getContent()); + LOG.error("Unable to fetch Kafka Supervisor status [{}] full response [{}]", + response.getStatusCode(), response.getContent()); return null; } } catch (Exception e) { @@ -556,11 +536,11 @@ private void checkLoadStatus(List segments) { String coordinatorResponse; try { - coordinatorResponse = - RetryUtils.retry(() -> DruidStorageHandlerUtils.getResponseFromCurrentLeader(getHttpClient(), - new Request(HttpMethod.GET, new URL(String.format("http://%s/status", coordinatorAddress))), - new StringFullResponseHandler(Charset.forName("UTF-8"))).getContent(), - input -> input instanceof IOException, maxTries); + coordinatorResponse = RetryUtils.retry(() -> + DruidStorageHandlerUtils.getResponseFromCurrentLeader(getHttpClient(), + new HiveDruidHttpRequest("GET", new URL(String.format("http://%s/status", coordinatorAddress)))) + .getContent(), + input -> input instanceof IOException, maxTries); } catch (Exception e) { CONSOLE.printInfo("Will skip waiting for data loading, coordinator unavailable"); return; @@ -591,12 +571,12 @@ private void checkLoadStatus(List segments) { urlsOfUnloadedSegments = ImmutableSet.copyOf(Sets.filter(urlsOfUnloadedSegments, input -> { try { String result = DruidStorageHandlerUtils - .getResponseFromCurrentLeader(getHttpClient(), new Request(HttpMethod.GET, input), - new StringFullResponseHandler(Charset.forName("UTF-8"))).getContent(); + .getResponseFromCurrentLeader(getHttpClient(), new HiveDruidHttpRequest("GET", input)) + .getContent(); LOG.debug("Checking segment [{}] response is [{}]", input, result); return Strings.isNullOrEmpty(result); - } catch (InterruptedException | ExecutionException e) { + } catch (IOException e) { LOG.error(String.format("Error while checking URL [%s]", input), e); return true; } @@ -884,7 +864,7 @@ private SQLMetadataConnector buildConnector() { connector = new MySQLConnector(storageConnectorConfigSupplier, Suppliers.ofInstance(getDruidMetadataStorageTablesConfig()), - new MySQLConnectorConfig()); + new MySQLConnectorSslConfig(), new MySQLConnectorDriverConfig()); break; case "postgresql": connector = @@ -934,33 +914,17 @@ private String getRootWorkingDir() { return rootWorkingDir; } - private static HttpClient makeHttpClient(Lifecycle lifecycle) { - final int - numConnection = - HiveConf.getIntVar(SessionState.getSessionConf(), HiveConf.ConfVars.HIVE_DRUID_NUM_HTTP_CONNECTION); - final Period - readTimeout = - new Period(HiveConf.getVar(SessionState.getSessionConf(), HiveConf.ConfVars.HIVE_DRUID_HTTP_READ_TIMEOUT)); - LOG.info("Creating Druid HTTP client with {} max parallel connections and {}ms read timeout", - numConnection, - readTimeout.toStandardDuration().getMillis()); - - final HttpClient - httpClient = - HttpClientInit.createClient(HttpClientConfig.builder() - .withNumConnections(numConnection) - .withReadTimeout(new Period(readTimeout).toStandardDuration()) - .build(), lifecycle); + private static HiveDruidHttpClient makeHttpClient() { + final Period readTimeout = new Period( + HiveConf.getVar(SessionState.getSessionConf(), HiveConf.ConfVars.HIVE_DRUID_HTTP_READ_TIMEOUT)); final boolean kerberosEnabled = HiveConf.getBoolVar(SessionState.getSessionConf(), HiveConf.ConfVars.HIVE_DRUID_KERBEROS_ENABLE); - if (kerberosEnabled && UserGroupInformation.isSecurityEnabled()) { - LOG.info("building Kerberos Http Client"); - return new KerberosHttpClient(httpClient); - } - return httpClient; + LOG.info("Creating HiveDruidHttpClient with {}ms read timeout, kerberos={}", + readTimeout.toStandardDuration().getMillis(), kerberosEnabled); + return new HiveDruidHttpClient((int) readTimeout.toStandardDuration().getMillis(), kerberosEnabled); } - public static HttpClient getHttpClient() { + public static HiveDruidHttpClient getHttpClient() { return HTTP_CLIENT; } diff --git a/druid-handler/src/java/org/apache/hadoop/hive/druid/DruidStorageHandlerUtils.java b/druid-handler/src/java/org/apache/hadoop/hive/druid/DruidStorageHandlerUtils.java index 7e49b0b7c96e..5a9d26eb3842 100644 --- a/druid-handler/src/java/org/apache/hadoop/hive/druid/DruidStorageHandlerUtils.java +++ b/druid-handler/src/java/org/apache/hadoop/hive/druid/DruidStorageHandlerUtils.java @@ -45,11 +45,9 @@ import org.apache.druid.java.util.emitter.EmittingLogger; import org.apache.druid.java.util.emitter.core.NoopEmitter; import org.apache.druid.java.util.emitter.service.ServiceEmitter; -import org.apache.druid.java.util.http.client.HttpClient; -import org.apache.druid.java.util.http.client.Request; -import org.apache.druid.java.util.http.client.response.InputStreamResponseHandler; -import org.apache.druid.java.util.http.client.response.StringFullResponseHandler; -import org.apache.druid.java.util.http.client.response.StringFullResponseHolder; +import org.apache.hadoop.hive.druid.http.HiveDruidHttpClient; +import org.apache.hadoop.hive.druid.http.HiveDruidHttpRequest; +import org.apache.hadoop.hive.druid.http.HiveDruidHttpResponse; import org.apache.druid.math.expr.ExprMacroTable; import org.apache.druid.metadata.MetadataStorageTablesConfig; import org.apache.druid.metadata.SQLMetadataConnector; @@ -89,6 +87,7 @@ import org.apache.druid.segment.IndexSpec; import org.apache.druid.segment.VirtualColumn; import org.apache.druid.segment.VirtualColumns; +import org.apache.druid.segment.column.ColumnType; import org.apache.druid.segment.column.ValueType; import org.apache.druid.segment.data.BitmapSerdeFactory; import org.apache.druid.segment.data.ConciseBitmapSerdeFactory; @@ -144,9 +143,6 @@ import org.apache.hadoop.io.BytesWritable; import org.apache.hadoop.io.retry.RetryPolicies; import org.apache.hadoop.io.retry.RetryProxy; -import org.jboss.netty.handler.codec.http.HttpHeaders; -import org.jboss.netty.handler.codec.http.HttpMethod; -import org.jboss.netty.handler.codec.http.HttpResponseStatus; import org.joda.time.DateTime; import org.joda.time.Interval; import org.joda.time.Period; @@ -217,7 +213,7 @@ private DruidStorageHandlerUtils() { /** * Mapper to use to serialize/deserialize Druid objects (SMILE). */ - public static final ObjectMapper SMILE_MAPPER = new DefaultObjectMapper(new SmileFactory()); + public static final ObjectMapper SMILE_MAPPER = new DefaultObjectMapper(new SmileFactory(), "smile"); private static final int DEFAULT_MAX_TRIES = 10; static { @@ -289,10 +285,11 @@ private DruidStorageHandlerUtils() { * @param query druid query. * @return Request object to be submitted. */ - public static Request createSmileRequest(String address, org.apache.druid.query.Query query) { + public static HiveDruidHttpRequest createSmileRequest(String address, org.apache.druid.query.Query query) { try { - return new Request(HttpMethod.POST, new URL(String.format("%s/druid/v2/", "http://" + address))).setContent( - SMILE_MAPPER.writeValueAsBytes(query)).setHeader(HttpHeaders.Names.CONTENT_TYPE, SMILE_CONTENT_TYPE); + return new HiveDruidHttpRequest("POST", new URL(String.format("%s/druid/v2/", "http://" + address))) + .setContent(SMILE_MAPPER.writeValueAsBytes(query)) + .setHeader("Content-Type", SMILE_CONTENT_TYPE); } catch (MalformedURLException e) { LOG.error("URL Malformed address {}", address); throw new RuntimeException(e); @@ -311,44 +308,30 @@ public static Request createSmileRequest(String address, org.apache.druid.query. * @return response object. * @throws IOException in case of request IO error. */ - public static InputStream submitRequest(HttpClient client, Request request) throws IOException { - try { - return client.go(request, new InputStreamResponseHandler()).get(); - } catch (ExecutionException | InterruptedException e) { - throw new IOException(e.getCause()); - } - + public static InputStream submitRequest(HiveDruidHttpClient client, HiveDruidHttpRequest request) + throws IOException { + return client.executeStream(request); } - static StringFullResponseHolder getResponseFromCurrentLeader(HttpClient client, Request request, - StringFullResponseHandler fullResponseHandler) throws ExecutionException, InterruptedException { - StringFullResponseHolder responseHolder = client.go(request, fullResponseHandler).get(); - if (HttpResponseStatus.TEMPORARY_REDIRECT.equals(responseHolder.getStatus())) { - String redirectUrlStr = responseHolder.getResponse().headers().get("Location"); - LOG.debug("Request[%s] received redirect response to location [%s].", request.getUrl(), redirectUrlStr); + static HiveDruidHttpResponse getResponseFromCurrentLeader(HiveDruidHttpClient client, + HiveDruidHttpRequest request) throws IOException { + HiveDruidHttpResponse responseHolder = client.execute(request); + if (responseHolder.getStatusCode() == HiveDruidHttpResponse.SC_TEMPORARY_REDIRECT) { + String redirectUrlStr = responseHolder.getHeader("Location"); + LOG.debug("Request[{}] received redirect response to location [{}].", request.getUrl(), redirectUrlStr); final URL redirectUrl; try { redirectUrl = new URL(redirectUrlStr); } catch (MalformedURLException ex) { - throw new ExecutionException(String - .format("Malformed redirect location is found in response from url[%s], new location[%s].", - request.getUrl(), - redirectUrlStr), ex); + throw new IOException(String.format( + "Malformed redirect location is found in response from url[%s], new location[%s].", + request.getUrl(), redirectUrlStr), ex); } - responseHolder = client.go(withUrl(request, redirectUrl), fullResponseHandler).get(); + responseHolder = client.execute(request.withUrl(redirectUrl)); } return responseHolder; } - private static Request withUrl(Request old, URL url) { - Request req = new Request(old.getMethod(), url); - req.addHeaderValues(old.getHeaders()); - if (old.hasContent()) { - req.setContent(old.getContent()); - } - return req; - } - /** * @param taskDir path to the directory containing the segments descriptor info * the descriptor path will be @@ -797,7 +780,7 @@ private static ShardSpec getNextPartitionShardSpec(ShardSpec shardSpec) { if (shardSpec instanceof LinearShardSpec) { return new LinearShardSpec(shardSpec.getPartitionNum() + 1); } else if (shardSpec instanceof NumberedShardSpec) { - return new NumberedShardSpec(shardSpec.getPartitionNum(), ((NumberedShardSpec) shardSpec).getPartitions()); + return new NumberedShardSpec(shardSpec.getPartitionNum(), ((NumberedShardSpec) shardSpec).getNumCorePartitions()); } else { // Druid only support appending more partitions to Linear and Numbered ShardSpecs. throw new IllegalStateException(String.format("Cannot expand shard spec [%s]", shardSpec)); @@ -832,12 +815,9 @@ public static IndexSpec getIndexSpec(Configuration jc) { if ("concise".equals(HiveConf.getVar(jc, HiveConf.ConfVars.HIVE_DRUID_BITMAP_FACTORY_TYPE))) { bitmapSerdeFactory = new ConciseBitmapSerdeFactory(); } else { - bitmapSerdeFactory = new RoaringBitmapSerdeFactory(true); + bitmapSerdeFactory = RoaringBitmapSerdeFactory.getInstance(); } - return new IndexSpec(bitmapSerdeFactory, - IndexSpec.DEFAULT_DIMENSION_COMPRESSION, - IndexSpec.DEFAULT_METRIC_COMPRESSION, - IndexSpec.DEFAULT_LONG_ENCODING); + return IndexSpec.builder().withBitmapSerdeFactory(bitmapSerdeFactory).build(); } public static Pair, AggregatorFactory[]> getDimensionsAndAggregates(List columnNames, @@ -1082,7 +1062,8 @@ private static BloomKFilter evaluateBloomFilter(ExprNodeDesc desc, Configuration Set usedColumnNames = virtualColumns.stream().map(col -> col.getOutputName()).collect(Collectors.toSet()); final String name = SqlValidatorUtil.uniquify("vc", usedColumnNames, SqlValidatorUtil.EXPR_SUGGESTER); ExpressionVirtualColumn expressionVirtualColumn = - new ExpressionVirtualColumn(name, virtualColumnExpr, targetType, ExprMacroTable.nil()); + new ExpressionVirtualColumn(name, virtualColumnExpr, ColumnType.fromString(targetType.toString()), + ExprMacroTable.nil()); virtualColumns.add(expressionVirtualColumn); return name; } diff --git a/druid-handler/src/java/org/apache/hadoop/hive/druid/http/HiveDruidHttpClient.java b/druid-handler/src/java/org/apache/hadoop/hive/druid/http/HiveDruidHttpClient.java new file mode 100644 index 000000000000..283ac3a37c25 --- /dev/null +++ b/druid-handler/src/java/org/apache/hadoop/hive/druid/http/HiveDruidHttpClient.java @@ -0,0 +1,211 @@ +/* + * 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.hadoop.hive.druid.http; + +import org.apache.hadoop.hive.druid.security.DruidKerberosUtil; +import org.apache.hadoop.security.UserGroupInformation; +import org.apache.http.client.config.RequestConfig; +import org.apache.http.client.methods.CloseableHttpResponse; +import org.apache.http.client.methods.HttpEntityEnclosingRequestBase; +import org.apache.http.client.methods.HttpGet; +import org.apache.http.client.methods.HttpPost; +import org.apache.http.client.methods.HttpRequestBase; +import org.apache.http.client.methods.HttpUriRequest; +import org.apache.http.entity.ByteArrayEntity; +import org.apache.http.impl.client.CloseableHttpClient; +import org.apache.http.impl.client.HttpClients; +import org.apache.http.util.EntityUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.Closeable; +import java.io.IOException; +import java.io.InputStream; +import java.net.CookieManager; +import java.net.HttpCookie; +import java.net.URI; +import java.security.PrivilegedExceptionAction; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; + +/** + * Apache HttpClient based HTTP client for Druid REST calls. + * Replaces Druid's Netty 3 HttpClientInit/KerberosHttpClient stack (HIVE-25013). + */ +public class HiveDruidHttpClient implements Closeable { + private static final Logger LOG = LoggerFactory.getLogger(HiveDruidHttpClient.class); + + private final CloseableHttpClient httpClient; + private final CookieManager cookieManager; + private final boolean kerberosEnabled; + private final ExecutorService executor; + + public HiveDruidHttpClient(int readTimeoutMs, boolean kerberosEnabled) { + this.kerberosEnabled = kerberosEnabled && UserGroupInformation.isSecurityEnabled(); + this.cookieManager = new CookieManager(); + this.httpClient = HttpClients.custom() + .setDefaultRequestConfig(RequestConfig.custom() + .setConnectTimeout(readTimeoutMs) + .setSocketTimeout(readTimeoutMs) + .setConnectionRequestTimeout(readTimeoutMs) + .build()) + .disableCookieManagement() + .build(); + this.executor = Executors.newCachedThreadPool(r -> { + Thread t = new Thread(r, "HiveDruidHttpClient"); + t.setDaemon(true); + return t; + }); + if (this.kerberosEnabled) { + LOG.info("HiveDruidHttpClient Kerberos authentication enabled"); + } + } + + public HiveDruidHttpResponse execute(HiveDruidHttpRequest request) throws IOException { + return innerExecute(request); + } + + public InputStream executeStream(HiveDruidHttpRequest request) throws IOException { + HiveDruidHttpResponse response = innerExecute(request); + return new java.io.ByteArrayInputStream(response.getBody()); + } + + public Future executeStreamAsync(HiveDruidHttpRequest request) { + return executor.submit(() -> executeStream(request)); + } + + private HiveDruidHttpResponse innerExecute(HiveDruidHttpRequest request) throws IOException { + HiveDruidHttpRequest current = request.copy(); + boolean shouldRetryOnUnauthorized = prepareKerberosAuth(current); + + while (true) { + try (CloseableHttpResponse response = httpClient.execute(buildHttpRequest(current))) { + int statusCode = response.getStatusLine().getStatusCode(); + storeCookies(current.getUrl().toURI(), response); + byte[] body = EntityUtils.toByteArray(response.getEntity()); + + if (kerberosEnabled && shouldRetryOnUnauthorized + && statusCode == HiveDruidHttpResponse.SC_UNAUTHORIZED) { + LOG.debug("Received 401 for URI {}, retrying with fresh Kerberos credentials", current.getUrl()); + DruidKerberosUtil.removeAuthCookie(cookieManager.getCookieStore(), current.getUrl().toURI()); + current = request.copy(); + current.setHeader("Cookie", ""); + shouldRetryOnUnauthorized = prepareKerberosAuth(current); + continue; + } + + return new HiveDruidHttpResponse(statusCode, body, extractHeaders(response)); + } catch (IOException e) { + throw e; + } catch (Exception e) { + throw new IOException(e); + } + } + } + + private boolean prepareKerberosAuth(HiveDruidHttpRequest request) throws IOException { + if (!kerberosEnabled) { + return false; + } + try { + URI uri = request.getUrl().toURI(); + Map> cookieHeaders = cookieManager.get(uri, Collections.emptyMap()); + for (Map.Entry> entry : cookieHeaders.entrySet()) { + request.addHeaderValues(entry.getKey(), entry.getValue()); + } + + if (DruidKerberosUtil.needToSendCredentials(cookieManager.getCookieStore(), uri)) { + UserGroupInformation currentUser = UserGroupInformation.getCurrentUser(); + currentUser.checkTGTAndReloginFromKeytab(); + String challenge = currentUser.doAs((PrivilegedExceptionAction) () -> + DruidKerberosUtil.kerberosChallenge(request.getUrl().getHost())); + request.setHeader("Authorization", "Negotiate " + challenge); + return false; + } + return true; + } catch (Exception e) { + throw new IOException("Failed to prepare Kerberos authentication", e); + } + } + + private static HttpUriRequest buildHttpRequest(HiveDruidHttpRequest request) throws IOException { + HttpRequestBase httpRequest; + if ("POST".equalsIgnoreCase(request.getMethod())) { + HttpPost post = new HttpPost(request.getUrl().toString()); + if (request.hasContent()) { + post.setEntity(new ByteArrayEntity(request.getContent())); + } + httpRequest = post; + } else if ("GET".equalsIgnoreCase(request.getMethod())) { + httpRequest = new HttpGet(request.getUrl().toString()); + } else { + HttpEntityEnclosingRequestBase custom = new HttpEntityEnclosingRequestBase() { + @Override + public String getMethod() { + return request.getMethod(); + } + }; + custom.setURI(URI.create(request.getUrl().toString())); + if (request.hasContent()) { + custom.setEntity(new ByteArrayEntity(request.getContent())); + } + httpRequest = custom; + } + for (Map.Entry> header : request.getHeaders().entrySet()) { + for (String value : header.getValue()) { + httpRequest.addHeader(header.getKey(), value); + } + } + return httpRequest; + } + + private static Map> extractHeaders(CloseableHttpResponse response) { + Map> headers = new HashMap<>(); + for (org.apache.http.Header header : response.getAllHeaders()) { + headers.computeIfAbsent(header.getName(), k -> new ArrayList<>()).add(header.getValue()); + } + return headers; + } + + private void storeCookies(URI uri, CloseableHttpResponse response) { + Map> headerMap = new HashMap<>(); + for (org.apache.http.Header header : response.getHeaders("Set-Cookie")) { + headerMap.computeIfAbsent("Set-Cookie", k -> new ArrayList<>()).add(header.getValue()); + } + if (!headerMap.isEmpty()) { + try { + cookieManager.put(uri, headerMap); + } catch (IOException e) { + LOG.warn("Failed to store cookies for URI {}", uri, e); + } + } + } + + @Override + public void close() throws IOException { + executor.shutdownNow(); + httpClient.close(); + } +} diff --git a/druid-handler/src/java/org/apache/hadoop/hive/druid/http/HiveDruidHttpRequest.java b/druid-handler/src/java/org/apache/hadoop/hive/druid/http/HiveDruidHttpRequest.java new file mode 100644 index 000000000000..ab6e0c04caad --- /dev/null +++ b/druid-handler/src/java/org/apache/hadoop/hive/druid/http/HiveDruidHttpRequest.java @@ -0,0 +1,104 @@ +/* + * 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.hadoop.hive.druid.http; + +import java.net.URL; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** + * HTTP request for Druid REST calls. Replaces Druid's Netty 3 based Request (HIVE-25013). + */ +public class HiveDruidHttpRequest { + private final String method; + private final URL url; + private final Map> headers = new HashMap<>(); + private byte[] content; + + public HiveDruidHttpRequest(String method, URL url) { + this.method = method; + this.url = url; + } + + public String getMethod() { + return method; + } + + public URL getUrl() { + return url; + } + + public byte[] getContent() { + return content; + } + + public boolean hasContent() { + return content != null && content.length > 0; + } + + public Map> getHeaders() { + return headers; + } + + public HiveDruidHttpRequest setContent(byte[] content) { + this.content = content; + return this; + } + + public HiveDruidHttpRequest setHeader(String name, String value) { + List values = new ArrayList<>(1); + values.add(value); + headers.put(name, values); + return this; + } + + public HiveDruidHttpRequest addHeaderValues(String name, List values) { + headers.put(name, new ArrayList<>(values)); + return this; + } + + public HiveDruidHttpRequest copy() { + HiveDruidHttpRequest copy = new HiveDruidHttpRequest(method, url); + for (Map.Entry> entry : headers.entrySet()) { + copy.headers.put(entry.getKey(), new ArrayList<>(entry.getValue())); + } + if (content != null) { + copy.content = content.clone(); + } + return copy; + } + + public HiveDruidHttpRequest withUrl(URL newUrl) { + HiveDruidHttpRequest copy = new HiveDruidHttpRequest(method, newUrl); + for (Map.Entry> entry : headers.entrySet()) { + copy.headers.put(entry.getKey(), new ArrayList<>(entry.getValue())); + } + if (content != null) { + copy.content = content.clone(); + } + return copy; + } + + public HiveDruidHttpRequest setContent(String content) { + return setContent(content.getBytes(StandardCharsets.UTF_8)); + } +} diff --git a/druid-handler/src/java/org/apache/hadoop/hive/druid/http/HiveDruidHttpResponse.java b/druid-handler/src/java/org/apache/hadoop/hive/druid/http/HiveDruidHttpResponse.java new file mode 100644 index 000000000000..d4219426a412 --- /dev/null +++ b/druid-handler/src/java/org/apache/hadoop/hive/druid/http/HiveDruidHttpResponse.java @@ -0,0 +1,75 @@ +/* + * 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.hadoop.hive.druid.http; + +import java.nio.charset.Charset; +import java.nio.charset.StandardCharsets; +import java.util.Collections; +import java.util.List; +import java.util.Map; + +/** + * HTTP response wrapper for Druid REST calls. Replaces Druid's Netty 3 based response holders (HIVE-25013). + */ +public class HiveDruidHttpResponse { + public static final int SC_OK = 200; + public static final int SC_BAD_REQUEST = 400; + public static final int SC_UNAUTHORIZED = 401; + public static final int SC_NOT_FOUND = 404; + public static final int SC_TEMPORARY_REDIRECT = 307; + + private final int statusCode; + private final byte[] body; + private final Map> headers; + + public HiveDruidHttpResponse(int statusCode, byte[] body, Map> headers) { + this.statusCode = statusCode; + this.body = body == null ? new byte[0] : body; + this.headers = headers == null ? Collections.emptyMap() : headers; + } + + public int getStatusCode() { + return statusCode; + } + + public byte[] getBody() { + return body; + } + + public String getContent() { + return getContent(StandardCharsets.UTF_8); + } + + public String getContent(Charset charset) { + return new String(body, charset); + } + + public String getHeader(String name) { + for (Map.Entry> entry : headers.entrySet()) { + if (entry.getKey().equalsIgnoreCase(name) && !entry.getValue().isEmpty()) { + return entry.getValue().get(0); + } + } + return null; + } + + public Map> getHeaders() { + return headers; + } +} diff --git a/druid-handler/src/java/org/apache/hadoop/hive/druid/io/DruidOutputFormat.java b/druid-handler/src/java/org/apache/hadoop/hive/druid/io/DruidOutputFormat.java index d90db9cbda92..f8ebc6b46eab 100644 --- a/druid-handler/src/java/org/apache/hadoop/hive/druid/io/DruidOutputFormat.java +++ b/druid-handler/src/java/org/apache/hadoop/hive/druid/io/DruidOutputFormat.java @@ -121,11 +121,12 @@ public FileSinkOperator.RecordWriter getHiveRecordWriter( .getDimensionsAndAggregates(columnNames, columnTypes); final InputRowParser inputRowParser = new MapInputRowParser(new TimeAndDimsParseSpec( new TimestampSpec(DruidConstants.DEFAULT_TIMESTAMP_COLUMN, "auto", null), - new DimensionsSpec(dimensionsAndAggregates.lhs, Lists - .newArrayList(Constants.DRUID_TIMESTAMP_GRANULARITY_COL_NAME, - Constants.DRUID_SHARD_KEY_COL_NAME - ), null - ) + DimensionsSpec.builder() + .setDimensions(dimensionsAndAggregates.lhs) + .setDimensionExclusions(Lists.newArrayList( + Constants.DRUID_TIMESTAMP_GRANULARITY_COL_NAME, + Constants.DRUID_SHARD_KEY_COL_NAME)) + .build() )); Map @@ -152,9 +153,16 @@ public FileSinkOperator.RecordWriter getHiveRecordWriter( Integer maxRowInMemory = HiveConf.getIntVar(jc, HiveConf.ConfVars.HIVE_DRUID_MAX_ROW_IN_MEMORY); IndexSpec indexSpec = DruidStorageHandlerUtils.getIndexSpec(jc); - RealtimeTuningConfig realtimeTuningConfig = new RealtimeTuningConfig(maxRowInMemory, null, null, null, - new File(basePersistDirectory, dataSource), new CustomVersioningPolicy(version), null, null, null, indexSpec, - null, true, 0, 0, true, null, 0L, null, null); + RealtimeTuningConfig defaults = + RealtimeTuningConfig.makeDefaultTuningConfig(new File(basePersistDirectory, dataSource)); + RealtimeTuningConfig realtimeTuningConfig = new RealtimeTuningConfig(defaults.getAppendableIndexSpec(), + maxRowInMemory, defaults.getMaxBytesInMemory(), defaults.isSkipBytesInMemoryOverheadCheck(), + defaults.getIntermediatePersistPeriod(), defaults.getWindowPeriod(), + new File(basePersistDirectory, dataSource), new CustomVersioningPolicy(version), + defaults.getRejectionPolicyFactory(), defaults.getMaxPendingPersists(), defaults.getShardSpec(), indexSpec, + indexSpec, defaults.getPersistThreadPriority(), defaults.getMergeThreadPriority(), + defaults.isReportParseExceptions(), defaults.getHandoffConditionTimeout(), defaults.getAlertTimeout(), + defaults.getSegmentWriteOutMediumFactory(), defaults.getDedupColumn()); LOG.debug(String.format("running with Data schema [%s] ", dataSchema)); return new DruidRecordWriter(dataSchema, realtimeTuningConfig, diff --git a/druid-handler/src/java/org/apache/hadoop/hive/druid/io/DruidQueryBasedInputFormat.java b/druid-handler/src/java/org/apache/hadoop/hive/druid/io/DruidQueryBasedInputFormat.java index 2a2be067125f..b20aba9ceaa6 100644 --- a/druid-handler/src/java/org/apache/hadoop/hive/druid/io/DruidQueryBasedInputFormat.java +++ b/druid-handler/src/java/org/apache/hadoop/hive/druid/io/DruidQueryBasedInputFormat.java @@ -21,7 +21,6 @@ import com.google.common.collect.Lists; import org.apache.commons.lang3.StringEscapeUtils; import org.apache.commons.lang3.StringUtils; -import org.apache.druid.java.util.http.client.Request; import org.apache.druid.query.BaseQuery; import org.apache.druid.query.LocatedSegmentDescriptor; import org.apache.druid.query.Query; @@ -53,8 +52,8 @@ import org.apache.hadoop.mapreduce.JobContext; import org.apache.hadoop.mapreduce.RecordReader; import org.apache.hadoop.mapreduce.TaskAttemptContext; +import org.apache.hadoop.hive.druid.http.HiveDruidHttpRequest; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; -import org.jboss.netty.handler.codec.http.HttpMethod; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -198,14 +197,14 @@ private static List fetchLocatedSegmentDescriptors(Str request = String.format("http://%s/druid/v2/datasources/%s/candidates?intervals=%s", address, - query.getDataSource().getNames().get(0), + query.getDataSource().getTableNames().iterator().next(), URLEncoder.encode(intervals, "UTF-8")); LOG.debug("sending request {} to query for segments", request); final InputStream response; try { response = DruidStorageHandlerUtils.submitRequest(DruidStorageHandler.getHttpClient(), - new Request(HttpMethod.GET, new URL(request))); + new HiveDruidHttpRequest("GET", new URL(request))); } catch (Exception e) { throw new IOException(org.apache.hadoop.util.StringUtils.stringifyException(e)); } diff --git a/druid-handler/src/java/org/apache/hadoop/hive/druid/io/DruidRecordWriter.java b/druid-handler/src/java/org/apache/hadoop/hive/druid/io/DruidRecordWriter.java index dc16c4e3f791..6eee0ec265f1 100644 --- a/druid-handler/src/java/org/apache/hadoop/hive/druid/io/DruidRecordWriter.java +++ b/druid-handler/src/java/org/apache/hadoop/hive/druid/io/DruidRecordWriter.java @@ -36,7 +36,9 @@ import org.apache.druid.segment.realtime.appenderator.Appenderators; import org.apache.druid.segment.realtime.appenderator.SegmentIdWithShardSpec; import org.apache.druid.segment.realtime.appenderator.SegmentNotWritableException; -import org.apache.druid.segment.realtime.appenderator.SegmentsAndMetadata; +import org.apache.druid.segment.incremental.ParseExceptionHandler; +import org.apache.druid.segment.incremental.SimpleRowIngestionMeters; +import org.apache.druid.segment.realtime.appenderator.SegmentsAndCommitMetadata; import org.apache.druid.segment.realtime.plumber.Committers; import org.apache.druid.timeline.DataSegment; import org.apache.druid.timeline.partition.LinearShardSpec; @@ -103,10 +105,11 @@ public DruidRecordWriter(DataSchema dataSchema, "realtimeTuningConfig is null"); this.dataSchema = Preconditions.checkNotNull(dataSchema, "data schema is null"); - appenderator = Appenderators - .createOffline("hive-offline-appenderator", this.dataSchema, tuningConfig, false, new FireDepartmentMetrics(), - dataSegmentPusher, DruidStorageHandlerUtils.JSON_MAPPER, DruidStorageHandlerUtils.INDEX_IO, - DruidStorageHandlerUtils.INDEX_MERGER_V9); + SimpleRowIngestionMeters rowIngestionMeters = new SimpleRowIngestionMeters(); + appenderator = Appenderators.createOffline("hive-offline-appenderator", this.dataSchema, tuningConfig, + new FireDepartmentMetrics(), dataSegmentPusher, DruidStorageHandlerUtils.JSON_MAPPER, + DruidStorageHandlerUtils.INDEX_IO, DruidStorageHandlerUtils.INDEX_MERGER_V9, rowIngestionMeters, + new ParseExceptionHandler(rowIngestionMeters, tuningConfig.isReportParseExceptions(), 0, 0), false); this.maxPartitionSize = maxPartitionSize; appenderator.startJob(); this.segmentsDescriptorDir = Preconditions.checkNotNull(segmentsDescriptorsDir, "segmentsDescriptorsDir is null"); @@ -170,7 +173,7 @@ private SegmentIdWithShardSpec getSegmentIdentifierAndMaybePush(long truncatedTi private void pushSegments(List segmentsToPush) { try { - SegmentsAndMetadata segmentsAndMetadata = appenderator.push(segmentsToPush, committerSupplier.get(), false).get(); + SegmentsAndCommitMetadata segmentsAndMetadata = appenderator.push(segmentsToPush, committerSupplier.get(), false).get(); final Set pushedSegmentIdentifierHashSet = new HashSet<>(); for (DataSegment pushedSegment : segmentsAndMetadata.getSegments()) { diff --git a/druid-handler/src/java/org/apache/hadoop/hive/druid/json/AvroParseSpec.java b/druid-handler/src/java/org/apache/hadoop/hive/druid/json/AvroParseSpec.java index 48d6cf2d6c8a..3ed817b48fed 100644 --- a/druid-handler/src/java/org/apache/hadoop/hive/druid/json/AvroParseSpec.java +++ b/druid-handler/src/java/org/apache/hadoop/hive/druid/json/AvroParseSpec.java @@ -41,7 +41,7 @@ public class AvroParseSpec extends ParseSpec { @JsonProperty("dimensionsSpec") DimensionsSpec dimensionsSpec, @JsonProperty("flattenSpec") JSONPathSpec flattenSpec) { super(timestampSpec != null ? timestampSpec : new TimestampSpec(null, null, null), - dimensionsSpec != null ? dimensionsSpec : new DimensionsSpec(null, null, null)); + dimensionsSpec != null ? dimensionsSpec : DimensionsSpec.EMPTY); this.flattenSpec = flattenSpec != null ? flattenSpec : JSONPathSpec.DEFAULT; } diff --git a/druid-handler/src/java/org/apache/hadoop/hive/druid/json/KafkaSupervisorTuningConfig.java b/druid-handler/src/java/org/apache/hadoop/hive/druid/json/KafkaSupervisorTuningConfig.java index 4e171612b39b..c9ede58b25b1 100644 --- a/druid-handler/src/java/org/apache/hadoop/hive/druid/json/KafkaSupervisorTuningConfig.java +++ b/druid-handler/src/java/org/apache/hadoop/hive/druid/json/KafkaSupervisorTuningConfig.java @@ -22,7 +22,6 @@ import com.fasterxml.jackson.annotation.JsonSubTypes; import com.fasterxml.jackson.annotation.JsonTypeInfo; import org.apache.druid.segment.IndexSpec; -import org.apache.druid.segment.indexing.TuningConfigs; import org.apache.druid.segment.writeout.SegmentWriteOutMediumFactory; import org.joda.time.Duration; import org.joda.time.Period; @@ -136,8 +135,8 @@ public Duration getOffsetFetchPeriod() { @Override public String toString() { return "KafkaSupervisorTuningConfig{" + "maxRowsInMemory=" + getMaxRowsInMemory() + ", maxRowsPerSegment=" - + getMaxRowsPerSegment() + ", maxTotalRows=" + getMaxTotalRows() + ", maxBytesInMemory=" + TuningConfigs - .getMaxBytesInMemoryOrDefault(getMaxBytesInMemory()) + ", intermediatePersistPeriod=" + + getMaxRowsPerSegment() + ", maxTotalRows=" + getMaxTotalRows() + ", maxBytesInMemory=" + + getMaxBytesInMemoryOrDefault() + ", intermediatePersistPeriod=" + getIntermediatePersistPeriod() + ", basePersistDirectory=" + getBasePersistDirectory() + ", maxPendingPersists=" + getMaxPendingPersists() + ", indexSpec=" + getIndexSpec() + ", reportParseExceptions=" + isReportParseExceptions() + ", handoffConditionTimeout=" diff --git a/druid-handler/src/java/org/apache/hadoop/hive/druid/json/SeekableStreamIndexTaskTuningConfig.java b/druid-handler/src/java/org/apache/hadoop/hive/druid/json/SeekableStreamIndexTaskTuningConfig.java index 289f0e8d4308..b858a75dacd2 100644 --- a/druid-handler/src/java/org/apache/hadoop/hive/druid/json/SeekableStreamIndexTaskTuningConfig.java +++ b/druid-handler/src/java/org/apache/hadoop/hive/druid/json/SeekableStreamIndexTaskTuningConfig.java @@ -19,6 +19,7 @@ package org.apache.hadoop.hive.druid.json; import com.fasterxml.jackson.annotation.JsonProperty; +import org.apache.druid.segment.incremental.AppendableIndexSpec; import org.apache.druid.indexer.partitions.DynamicPartitionsSpec; import org.apache.druid.segment.IndexSpec; import org.apache.druid.segment.indexing.RealtimeTuningConfig; @@ -39,8 +40,10 @@ public abstract class SeekableStreamIndexTaskTuningConfig implements TuningConfi private static final boolean DEFAULT_RESET_OFFSET_AUTOMATICALLY = false; private static final boolean DEFAULT_SKIP_SEQUENCE_NUMBER_AVAILABILITY_CHECK = false; + private final AppendableIndexSpec appendableIndexSpec; private final int maxRowsInMemory; private final long maxBytesInMemory; + private final boolean skipBytesInMemoryOverheadCheck; private final DynamicPartitionsSpec partitionsSpec; private final Period intermediatePersistPeriod; private final File basePersistDirectory; @@ -85,11 +88,13 @@ public SeekableStreamIndexTaskTuningConfig( // Cannot be a static because default basePersistDirectory is unique per-instance final RealtimeTuningConfig defaults = RealtimeTuningConfig.makeDefaultTuningConfig(basePersistDirectory); + this.appendableIndexSpec = defaults.getAppendableIndexSpec(); this.maxRowsInMemory = maxRowsInMemory == null ? defaults.getMaxRowsInMemory() : maxRowsInMemory; this.partitionsSpec = new DynamicPartitionsSpec(maxRowsPerSegment, maxTotalRows); // initializing this to 0, it will be lazily initialized to a value // @see server.src.main.java.org.apache.druid.segment.indexing.TuningConfigs#getMaxBytesInMemoryOrDefault(long) this.maxBytesInMemory = maxBytesInMemory == null ? 0 : maxBytesInMemory; + this.skipBytesInMemoryOverheadCheck = defaults.isSkipBytesInMemoryOverheadCheck(); this.intermediatePersistPeriod = intermediatePersistPeriod == null ? defaults.getIntermediatePersistPeriod() : intermediatePersistPeriod; @@ -131,6 +136,12 @@ public SeekableStreamIndexTaskTuningConfig( : logParseExceptions; } + @Override + @JsonProperty + public AppendableIndexSpec getAppendableIndexSpec() { + return appendableIndexSpec; + } + @Override @JsonProperty public int getMaxRowsInMemory() { @@ -143,6 +154,12 @@ public long getMaxBytesInMemory() { return maxBytesInMemory; } + @Override + @JsonProperty + public boolean isSkipBytesInMemoryOverheadCheck() { + return skipBytesInMemoryOverheadCheck; + } + @Override @JsonProperty public Integer getMaxRowsPerSegment() { diff --git a/druid-handler/src/java/org/apache/hadoop/hive/druid/security/DruidKerberosUtil.java b/druid-handler/src/java/org/apache/hadoop/hive/druid/security/DruidKerberosUtil.java index 12603c10ec7e..0ff5e92eae09 100644 --- a/druid-handler/src/java/org/apache/hadoop/hive/druid/security/DruidKerberosUtil.java +++ b/druid-handler/src/java/org/apache/hadoop/hive/druid/security/DruidKerberosUtil.java @@ -55,7 +55,7 @@ private DruidKerberosUtil() { * @throws AuthenticationException on authentication errors. */ - static String kerberosChallenge(String server) throws AuthenticationException { + public static String kerberosChallenge(String server) throws AuthenticationException { KERBEROS_LOCK.lock(); try { // This Oid for Kerberos GSS-API mechanism. @@ -84,7 +84,7 @@ static String kerberosChallenge(String server) throws AuthenticationException { } } - static HttpCookie getAuthCookie(CookieStore cookieStore, URI uri) { + public static HttpCookie getAuthCookie(CookieStore cookieStore, URI uri) { if (cookieStore == null) { return null; } @@ -105,14 +105,14 @@ static HttpCookie getAuthCookie(CookieStore cookieStore, URI uri) { return null; } - static void removeAuthCookie(CookieStore cookieStore, URI uri) { + public static void removeAuthCookie(CookieStore cookieStore, URI uri) { HttpCookie authCookie = getAuthCookie(cookieStore, uri); if (authCookie != null) { cookieStore.remove(uri, authCookie); } } - static boolean needToSendCredentials(CookieStore cookieStore, URI uri) { + public static boolean needToSendCredentials(CookieStore cookieStore, URI uri) { return getAuthCookie(cookieStore, uri) == null; } diff --git a/druid-handler/src/java/org/apache/hadoop/hive/druid/security/KerberosHttpClient.java b/druid-handler/src/java/org/apache/hadoop/hive/druid/security/KerberosHttpClient.java deleted file mode 100644 index fdbbfcc7d7f6..000000000000 --- a/druid-handler/src/java/org/apache/hadoop/hive/druid/security/KerberosHttpClient.java +++ /dev/null @@ -1,132 +0,0 @@ -/* - * 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.hadoop.hive.druid.security; - -import com.google.common.base.Throwables; -import com.google.common.util.concurrent.ListenableFuture; -import com.google.common.util.concurrent.SettableFuture; -import org.apache.druid.java.util.http.client.AbstractHttpClient; -import org.apache.druid.java.util.http.client.HttpClient; -import org.apache.druid.java.util.http.client.Request; -import org.apache.druid.java.util.http.client.response.HttpResponseHandler; -import org.apache.hadoop.security.UserGroupInformation; -import org.jboss.netty.handler.codec.http.HttpHeaders; -import org.joda.time.Duration; -import org.slf4j.LoggerFactory; - -import java.net.CookieManager; -import java.net.URI; -import java.security.PrivilegedExceptionAction; -import java.util.Collections; -import java.util.List; -import java.util.Map; - -/** - * This is a slightly modified version of kerberos module borrowed from druid project - * Couple of reasons behind the copy/modification instead of mvn dependency. - * 1/ Need to remove the authentication step since it not required - * 2/ To avoid some un-needed transitive dependencies that can clash on the classpath like jetty-XX. - */ -public class KerberosHttpClient extends AbstractHttpClient { - protected static final org.slf4j.Logger LOG = LoggerFactory.getLogger(KerberosHttpClient.class); - private final HttpClient delegate; - private final CookieManager cookieManager; - - public KerberosHttpClient(HttpClient delegate) { - this.delegate = delegate; - this.cookieManager = new CookieManager(); - } - - @Override public ListenableFuture go(Request request, - HttpResponseHandler httpResponseHandler, - Duration duration) { - final SettableFuture retVal = SettableFuture.create(); - innerGo(request, httpResponseHandler, duration, retVal); - return retVal; - } - - private void innerGo(final Request request, - final HttpResponseHandler httpResponseHandler, - final Duration duration, - final SettableFuture future) { - try { - final String host = request.getUrl().getHost(); - final URI uri = request.getUrl().toURI(); - - /* Cookies Manager is used to cache cookie returned by service. - The goal us to avoid doing KDC requests for every request.*/ - - Map> cookieMap = cookieManager.get(uri, Collections.emptyMap()); - for (Map.Entry> entry : cookieMap.entrySet()) { - request.addHeaderValues(entry.getKey(), entry.getValue()); - } - final boolean shouldRetryOnUnauthorizedResponse; - - if (DruidKerberosUtil.needToSendCredentials(cookieManager.getCookieStore(), uri)) { - // No Cookies for requested URI, authenticate user and add authentication header - LOG.debug("No Auth Cookie found for URI{}. Existing Cookies{} Authenticating... ", - uri, - cookieManager.getCookieStore().getCookies()); - // Assuming that a valid UGI with kerberos cred is created by HS2 or LLAP - UserGroupInformation currentUser = UserGroupInformation.getCurrentUser(); - currentUser.checkTGTAndReloginFromKeytab(); - LOG.debug("The user credential is {}", currentUser); - String - challenge = - currentUser.doAs((PrivilegedExceptionAction) () -> DruidKerberosUtil.kerberosChallenge(host)); - request.setHeader(HttpHeaders.Names.AUTHORIZATION, "Negotiate " + challenge); - /* no reason to retry if the challenge ticket is not valid. */ - shouldRetryOnUnauthorizedResponse = false; - } else { - /* In this branch we had already a cookie that did expire - therefore we need to resend a valid Kerberos challenge*/ - LOG.debug("Found Auth Cookie found for URI {} cookie {}", - uri, - DruidKerberosUtil.getAuthCookie(cookieManager.getCookieStore(), uri).toString()); - shouldRetryOnUnauthorizedResponse = true; - } - - @SuppressWarnings("unchecked") ListenableFuture> - internalFuture = - delegate.go(request, - new RetryIfUnauthorizedResponseHandler(new ResponseCookieHandler(request.getUrl() - .toURI(), cookieManager, httpResponseHandler)), - duration); - - RetryResponseHolder responseHolder = internalFuture.get(); - - if (shouldRetryOnUnauthorizedResponse && responseHolder.shouldRetry()) { - LOG.debug("Preparing for Retry boolean {} and result {}, object{} ", true, - responseHolder.shouldRetry(), - responseHolder.getObj()); - // remove Auth cookie - DruidKerberosUtil.removeAuthCookie(cookieManager.getCookieStore(), uri); - // clear existing cookie - request.setHeader("Cookie", ""); - innerGo(request.copy(), httpResponseHandler, duration, future); - - } else { - future.set(responseHolder.getObj()); - } - } catch (Throwable e) { - throw Throwables.propagate(e); - } - } - -} diff --git a/druid-handler/src/java/org/apache/hadoop/hive/druid/security/ResponseCookieHandler.java b/druid-handler/src/java/org/apache/hadoop/hive/druid/security/ResponseCookieHandler.java deleted file mode 100644 index 223000e62a04..000000000000 --- a/druid-handler/src/java/org/apache/hadoop/hive/druid/security/ResponseCookieHandler.java +++ /dev/null @@ -1,75 +0,0 @@ -/* - * 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.hadoop.hive.druid.security; - -import com.google.common.collect.Maps; -import org.apache.druid.java.util.http.client.response.ClientResponse; -import org.apache.druid.java.util.http.client.response.HttpResponseHandler; -import org.jboss.netty.handler.codec.http.HttpChunk; -import org.jboss.netty.handler.codec.http.HttpHeaders; -import org.jboss.netty.handler.codec.http.HttpResponse; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.io.IOException; -import java.net.CookieManager; -import java.net.URI; - -/** - * Class to handle Cookies used to cache the Kerberos Credentials. - * @param intermediate response type. - * @param final response type. - */ -public class ResponseCookieHandler implements HttpResponseHandler { - private static final Logger LOG = LoggerFactory.getLogger(ResponseCookieHandler.class); - - private final URI uri; - private final CookieManager manager; - private final HttpResponseHandler delegate; - - ResponseCookieHandler(URI uri, CookieManager manager, HttpResponseHandler delegate) { - this.uri = uri; - this.manager = manager; - this.delegate = delegate; - } - - @Override public ClientResponse handleResponse(HttpResponse httpResponse, TrafficCop trafficCop) { - try { - final HttpHeaders headers = httpResponse.headers(); - manager.put(uri, Maps.asMap(headers.names(), headers::getAll)); - return delegate.handleResponse(httpResponse, trafficCop); - } catch (IOException e) { - LOG.error("Error while processing Cookies from header", e); - throw new RuntimeException(e); - } - } - - @Override public ClientResponse handleChunk(ClientResponse clientResponse, - HttpChunk httpChunk, long chunkNum) { - return delegate.handleChunk(clientResponse, httpChunk, chunkNum); - } - - @Override public ClientResponse done(ClientResponse clientResponse) { - return delegate.done(clientResponse); - } - - @Override public void exceptionCaught(ClientResponse clientResponse, Throwable throwable) { - delegate.exceptionCaught(clientResponse, throwable); - } -} diff --git a/druid-handler/src/java/org/apache/hadoop/hive/druid/security/RetryIfUnauthorizedResponseHandler.java b/druid-handler/src/java/org/apache/hadoop/hive/druid/security/RetryIfUnauthorizedResponseHandler.java deleted file mode 100644 index d6702f5a1fe0..000000000000 --- a/druid-handler/src/java/org/apache/hadoop/hive/druid/security/RetryIfUnauthorizedResponseHandler.java +++ /dev/null @@ -1,101 +0,0 @@ -/* - * 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.hadoop.hive.druid.security; - -import org.apache.druid.java.util.http.client.response.ClientResponse; -import org.apache.druid.java.util.http.client.response.HttpResponseHandler; -import org.jboss.netty.handler.codec.http.HttpChunk; -import org.jboss.netty.handler.codec.http.HttpResponse; -import org.jboss.netty.handler.codec.http.HttpResponseStatus; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -/** - * Class handling retry on Unauthorized responses. - * - * @param Intermediate response type. - * @param final result type. - */ -public class RetryIfUnauthorizedResponseHandler - implements HttpResponseHandler, RetryResponseHolder> { - protected static final Logger LOG = LoggerFactory.getLogger(RetryIfUnauthorizedResponseHandler.class); - - private final HttpResponseHandler httpResponseHandler; - - public RetryIfUnauthorizedResponseHandler(HttpResponseHandler httpResponseHandler) { - this.httpResponseHandler = httpResponseHandler; - } - - @Override public ClientResponse> handleResponse(HttpResponse httpResponse, - TrafficCop trafficCop) { - LOG.debug("UnauthorizedResponseHandler - Got response status {}", httpResponse.getStatus()); - if (httpResponse.getStatus().equals(HttpResponseStatus.UNAUTHORIZED)) { - // Drain the buffer - //noinspection ResultOfMethodCallIgnored - httpResponse.getContent().toString(); - return ClientResponse.unfinished(RetryResponseHolder.retry()); - } else { - return wrap(httpResponseHandler.handleResponse(httpResponse, trafficCop)); - } - } - - @Override public ClientResponse> handleChunk( - ClientResponse> clientResponse, - HttpChunk httpChunk, long chunkNum) { - if (clientResponse.getObj().shouldRetry()) { - // Drain the buffer - //noinspection ResultOfMethodCallIgnored - httpChunk.getContent().toString(); - return clientResponse; - } else { - return wrap(httpResponseHandler.handleChunk(unwrap(clientResponse), httpChunk, chunkNum)); - } - } - - @Override public ClientResponse> done( - ClientResponse> clientResponse) { - if (clientResponse.getObj().shouldRetry()) { - return ClientResponse.finished(RetryResponseHolder.retry()); - } else { - return wrap(httpResponseHandler.done(unwrap(clientResponse))); - } - } - - @Override public void exceptionCaught(ClientResponse> clientResponse, - Throwable throwable) { - httpResponseHandler.exceptionCaught(unwrap(clientResponse), throwable); - } - - private ClientResponse> wrap(ClientResponse response) { - if (response.isFinished()) { - return ClientResponse.finished(new RetryResponseHolder<>(false, response.getObj())); - } else { - return ClientResponse.unfinished(new RetryResponseHolder<>(false, response.getObj())); - } - } - - private ClientResponse unwrap(ClientResponse> response) { - if (response.isFinished()) { - return ClientResponse.finished(response.getObj().getObj()); - } else { - return ClientResponse.unfinished(response.getObj().getObj()); - } - } - -} diff --git a/druid-handler/src/java/org/apache/hadoop/hive/druid/serde/DruidQueryRecordReader.java b/druid-handler/src/java/org/apache/hadoop/hive/druid/serde/DruidQueryRecordReader.java index 19379e1724da..c3d69b6e041b 100644 --- a/druid-handler/src/java/org/apache/hadoop/hive/druid/serde/DruidQueryRecordReader.java +++ b/druid-handler/src/java/org/apache/hadoop/hive/druid/serde/DruidQueryRecordReader.java @@ -26,10 +26,9 @@ import com.google.common.base.Throwables; import org.apache.druid.java.util.common.IAE; import org.apache.druid.java.util.common.RE; -import org.apache.druid.java.util.common.guava.CloseQuietly; -import org.apache.druid.java.util.http.client.HttpClient; -import org.apache.druid.java.util.http.client.Request; -import org.apache.druid.java.util.http.client.response.InputStreamResponseHandler; +import org.apache.commons.io.IOUtils; +import org.apache.hadoop.hive.druid.http.HiveDruidHttpClient; +import org.apache.hadoop.hive.druid.http.HiveDruidHttpRequest; import org.apache.druid.query.Query; import org.apache.druid.query.QueryInterruptedException; import org.apache.hadoop.conf.Configuration; @@ -73,7 +72,7 @@ public abstract class DruidQueryRecordReader> extends Re private ObjectMapper smileMapper; private Configuration conf; private String[] locations; - private HttpClient httpClient; + private HiveDruidHttpClient httpClient; /** * Query that Druid executes. */ @@ -116,8 +115,8 @@ public JsonParserIterator createQueryResultsIterator() { // Execute query LOG.debug("Retrieving data from druid location[{}] using query:[{}] ", address, query); try { - Request request = DruidStorageHandlerUtils.createSmileRequest(address, query); - Future inputStreamFuture = httpClient.go(request, new InputStreamResponseHandler()); + HiveDruidHttpRequest request = DruidStorageHandlerUtils.createSmileRequest(address, query); + Future inputStreamFuture = httpClient.executeStreamAsync(request); //noinspection unchecked iterator = new JsonParserIterator(smileMapper, resultsType, inputStreamFuture, request.getUrl().toString(), query); @@ -126,7 +125,7 @@ public JsonParserIterator createQueryResultsIterator() { } catch (Exception e) { if (iterator != null) { // We got exception while querying results from this host. - CloseQuietly.close(iterator); + IOUtils.closeQuietly(iterator); } LOG.error("Failure getting results for query[{}] from host[{}] because of [{}]", query, address, e.getMessage()); @@ -149,8 +148,8 @@ public JsonParserIterator createQueryResultsIterator() { initialize(split, context.getConfiguration()); } - public void initialize(InputSplit split, ObjectMapper mapper, ObjectMapper smileMapper, HttpClient httpClient, - Configuration conf) throws IOException { + public void initialize(InputSplit split, ObjectMapper mapper, ObjectMapper smileMapper, + HiveDruidHttpClient httpClient, Configuration conf) throws IOException { this.conf = conf; HiveDruidSplit hiveDruidSplit = (HiveDruidSplit) split; Preconditions.checkNotNull(hiveDruidSplit, "input split is null ???"); @@ -200,7 +199,7 @@ public void initialize(InputSplit split, Configuration conf) throws IOException @Override public void close() { if (queryResultsIterator != null) { - CloseQuietly.close(queryResultsIterator); + IOUtils.closeQuietly(queryResultsIterator); } } @@ -248,7 +247,7 @@ public void initialize(InputSplit split, Configuration conf) throws IOException return false; } if (jp.getCurrentToken() == JsonToken.END_ARRAY) { - CloseQuietly.close(jp); + IOUtils.closeQuietly(jp); return false; } @@ -296,7 +295,7 @@ private void init() { } @Override public void close() throws IOException { - CloseQuietly.close(jp); + IOUtils.closeQuietly(jp); } } diff --git a/druid-handler/src/test/org/apache/hadoop/hive/druid/serde/TestDruidSerDe.java b/druid-handler/src/test/org/apache/hadoop/hive/druid/serde/TestDruidSerDe.java index 5157e4656c4d..7f98d5b1e2e3 100644 --- a/druid-handler/src/test/org/apache/hadoop/hive/druid/serde/TestDruidSerDe.java +++ b/druid-handler/src/test/org/apache/hadoop/hive/druid/serde/TestDruidSerDe.java @@ -34,8 +34,8 @@ import java.util.Properties; import java.util.stream.Collectors; -import org.apache.druid.java.util.http.client.HttpClient; -import org.apache.druid.java.util.http.client.response.HttpResponseHandler; +import org.apache.hadoop.hive.druid.http.HiveDruidHttpClient; +import org.apache.hadoop.hive.druid.http.HiveDruidHttpRequest; import org.apache.druid.query.scan.ScanResultValue; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; @@ -660,10 +660,10 @@ private void deserializeQueryResults(DruidSerDe serDe, IllegalAccessException, InterruptedException, NoSuchMethodException, InvocationTargetException { // Initialize - HttpClient httpClient = mock(HttpClient.class); + HiveDruidHttpClient httpClient = mock(HiveDruidHttpClient.class); SettableFuture futureResult = SettableFuture.create(); futureResult.set(new ByteArrayInputStream(resultString)); - when(httpClient.go(any(), any(HttpResponseHandler.class))).thenReturn(futureResult); + when(httpClient.executeStreamAsync(any(HiveDruidHttpRequest.class))).thenReturn(futureResult); DruidQueryRecordReader reader = DruidQueryBasedInputFormat.getDruidQueryReader(queryType); final HiveDruidSplit split = new HiveDruidSplit(jsonQuery, new Path("empty"), new String[]{"testing_host"}); @@ -692,7 +692,7 @@ private void deserializeQueryResults(DruidSerDe serDe, // Check mapreduce path futureResult = SettableFuture.create(); futureResult.set(new ByteArrayInputStream(resultString)); - when(httpClient.go(any(), any(HttpResponseHandler.class))).thenReturn(futureResult); + when(httpClient.executeStreamAsync(any(HiveDruidHttpRequest.class))).thenReturn(futureResult); reader = DruidQueryBasedInputFormat.getDruidQueryReader(queryType); reader.initialize(split, DruidStorageHandlerUtils.JSON_MAPPER, DruidStorageHandlerUtils.SMILE_MAPPER, httpClient, conf); diff --git a/druid-handler/src/test/org/apache/hadoop/hive/ql/io/TestDruidRecordWriter.java b/druid-handler/src/test/org/apache/hadoop/hive/ql/io/TestDruidRecordWriter.java index 3fb7bdf6ca19..f2fba695ede1 100644 --- a/druid-handler/src/test/org/apache/hadoop/hive/ql/io/TestDruidRecordWriter.java +++ b/druid-handler/src/test/org/apache/hadoop/hive/ql/io/TestDruidRecordWriter.java @@ -130,7 +130,7 @@ inputRowParser = new MapInputRowParser(new TimeAndDimsParseSpec(new TimestampSpec(DruidConstants.DEFAULT_TIMESTAMP_COLUMN, "auto", - null), new DimensionsSpec(ImmutableList.of(new StringDimensionSchema("host")), null, null))); + null), new DimensionsSpec(ImmutableList.of(new StringDimensionSchema("host"))))); final Map parserMap = objectMapper.convertValue(inputRowParser, new TypeReference>() { @@ -146,12 +146,17 @@ null, objectMapper); - IndexSpec indexSpec = new IndexSpec(new RoaringBitmapSerdeFactory(true), null, null, null); - RealtimeTuningConfig - tuningConfig = - new RealtimeTuningConfig(null, - null, null, null, temporaryFolder.newFolder(), null, null, null, null, indexSpec, null, null, 0, 0, null, - null, 0L, null, null); + IndexSpec indexSpec = IndexSpec.builder().withBitmapSerdeFactory(RoaringBitmapSerdeFactory.getInstance()).build(); + RealtimeTuningConfig defaults = RealtimeTuningConfig.makeDefaultTuningConfig(temporaryFolder.newFolder()); + RealtimeTuningConfig tuningConfig = defaults.withBasePersistDirectory(temporaryFolder.newFolder()) + .withVersioningPolicy(defaults.getVersioningPolicy()); + tuningConfig = new RealtimeTuningConfig(defaults.getAppendableIndexSpec(), defaults.getMaxRowsInMemory(), + defaults.getMaxBytesInMemory(), defaults.isSkipBytesInMemoryOverheadCheck(), + defaults.getIntermediatePersistPeriod(), defaults.getWindowPeriod(), temporaryFolder.newFolder(), + defaults.getVersioningPolicy(), defaults.getRejectionPolicyFactory(), defaults.getMaxPendingPersists(), + defaults.getShardSpec(), indexSpec, indexSpec, defaults.getPersistThreadPriority(), + defaults.getMergeThreadPriority(), defaults.isReportParseExceptions(), defaults.getHandoffConditionTimeout(), + defaults.getAlertTimeout(), defaults.getSegmentWriteOutMediumFactory(), defaults.getDedupColumn()); LocalFileSystem localFileSystem = FileSystem.getLocal(config); DataSegmentPusher dataSegmentPusher = new LocalDataSegmentPusher(new LocalDataSegmentPusherConfig() { @Override public File getStorageDirectory() { diff --git a/packaging/pom.xml b/packaging/pom.xml index df4d33309e31..7996ec0222a2 100644 --- a/packaging/pom.xml +++ b/packaging/pom.xml @@ -279,6 +279,19 @@ kubernetes + + druid-handler + + + + org.apache.hive + hive-druid-handler + ${project.version} + + + @@ -353,11 +366,6 @@ hive-hbase-handler ${project.version} - - org.apache.hive - hive-druid-handler - ${project.version} - org.apache.hive hive-iceberg-handler diff --git a/pom.xml b/pom.xml index 4481607165db..4725b824f628 100644 --- a/pom.xml +++ b/pom.xml @@ -134,7 +134,7 @@ 10.17.1.0 3.1.0 0.1.2 - 0.17.1 + 26.0.0 2.2.4 1.12.0 22.0 @@ -156,7 +156,7 @@ 5.3.4 5.5 2.5.2 - 2.18.6 + 2.18.8 2.3.4 2.4.1 3.1.0 @@ -192,8 +192,8 @@ 5.17.0 5.2.0 2.0.0-M5 - 4.1.127.Final - 3.10.5.Final + 4.1.133.Final + 2.15.0 4.5.8 2.8 @@ -448,6 +448,11 @@ jjwt-jackson ${jjwt.version} + + org.asynchttpclient + async-http-client + ${async-http-client.version} + io.netty netty-all diff --git a/standalone-metastore/pom.xml b/standalone-metastore/pom.xml index 2eb75caf9281..144a85f58f98 100644 --- a/standalone-metastore/pom.xml +++ b/standalone-metastore/pom.xml @@ -87,7 +87,7 @@ 22.0 3.4.2 4.0.3 - 2.18.6 + 2.18.8 3.3 5.5.1 4.13.2 @@ -105,7 +105,7 @@ 0.10.2 1.72.0 1.9.0 - 4.1.127.Final + 4.1.133.Final 3.25.0 4.0.4 diff --git a/storage-api/pom.xml b/storage-api/pom.xml index 3627fc9cf270..29a43a537f5b 100644 --- a/storage-api/pom.xml +++ b/storage-api/pom.xml @@ -29,7 +29,7 @@ 11.1.0 21 21 - 4.1.127.Final + 4.1.133.Final 22.0 3.4.2 4.13.2