From 9ac4cca0ccc561f8e1bdc58ea31f7a39924612df Mon Sep 17 00:00:00 2001 From: Nikola Grcevski Date: Mon, 16 Dec 2019 16:46:40 -0500 Subject: [PATCH 1/3] Fix socket leak in benchmark client. --- .../org/apache/pinotdruidbenchmark/PinotThroughput.java | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/contrib/pinot-druid-benchmark/src/main/java/org/apache/pinotdruidbenchmark/PinotThroughput.java b/contrib/pinot-druid-benchmark/src/main/java/org/apache/pinotdruidbenchmark/PinotThroughput.java index 1ac1bf255ba3..3d7c3ee8c038 100644 --- a/contrib/pinot-druid-benchmark/src/main/java/org/apache/pinotdruidbenchmark/PinotThroughput.java +++ b/contrib/pinot-druid-benchmark/src/main/java/org/apache/pinotdruidbenchmark/PinotThroughput.java @@ -33,7 +33,7 @@ import org.apache.http.entity.StringEntity; import org.apache.http.impl.client.CloseableHttpClient; import org.apache.http.impl.client.HttpClients; - +import org.apache.http.util.EntityUtils; /** * Test throughput for Pinot. @@ -90,8 +90,9 @@ public void run() { try (CloseableHttpClient httpClient = HttpClients.createDefault()) { while (System.currentTimeMillis() < endTime) { long startTime = System.currentTimeMillis(); - CloseableHttpResponse httpResponse = httpClient.execute(httpPosts[RANDOM.nextInt(numQueries)]); - httpResponse.close(); + try (CloseableHttpResponse httpResponse = httpClient.execute(httpPosts[RANDOM.nextInt(numQueries)])) { + EntityUtils.consume(httpResponse.getEntity()); + } long responseTime = System.currentTimeMillis() - startTime; counter.getAndIncrement(); totalResponseTime.getAndAdd(responseTime); From 9c55f710f8547fa09505c3458b47fc27a47f051a Mon Sep 17 00:00:00 2001 From: Nikola Grcevski Date: Mon, 16 Dec 2019 16:47:02 -0500 Subject: [PATCH 2/3] Fix issue with benchmark server boot --- .../pinot/controller/helix/core/PinotHelixResourceManager.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java index 1477396337cd..886c2d2a73a8 100644 --- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java +++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java @@ -259,7 +259,7 @@ public ZkHelixPropertyStore getPropertyStore() { private void addInstanceGroupTagIfNeeded() { InstanceConfig instanceConfig = getHelixInstanceConfig(_instanceId); assert instanceConfig != null; - if (!instanceConfig.containsTag(Helix.CONTROLLER_INSTANCE)) { + if (instanceConfig != null && !instanceConfig.containsTag(Helix.CONTROLLER_INSTANCE)) { LOGGER.info("Controller: {} doesn't contain group tag: {}. Adding one.", _instanceId, Helix.CONTROLLER_INSTANCE); instanceConfig.addTag(Helix.CONTROLLER_INSTANCE); HelixDataAccessor accessor = _helixZkManager.getHelixDataAccessor(); From fd2330b3518704f74f8a761ec08bf95f739f5fec Mon Sep 17 00:00:00 2001 From: Nikola Grcevski Date: Tue, 17 Dec 2019 13:31:11 -0500 Subject: [PATCH 3/3] Reuse bytebuffer when possible Co-authored-by: Charlie Gracie --- .../java/org/apache/pinot/core/common/ObjectSerDeUtils.java | 3 +++ 1 file changed, 3 insertions(+) diff --git a/pinot-core/src/main/java/org/apache/pinot/core/common/ObjectSerDeUtils.java b/pinot-core/src/main/java/org/apache/pinot/core/common/ObjectSerDeUtils.java index 0754cd43f6c6..7b00f332f818 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/common/ObjectSerDeUtils.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/common/ObjectSerDeUtils.java @@ -140,6 +140,9 @@ public String deserialize(byte[] bytes) { @Override public String deserialize(ByteBuffer byteBuffer) { + if (byteBuffer.hasArray()) { + return StringUtil.decodeUtf8(byteBuffer.array(), byteBuffer.position() + byteBuffer.arrayOffset(), byteBuffer.remaining()); + } byte[] bytes = new byte[byteBuffer.remaining()]; byteBuffer.get(bytes); return StringUtil.decodeUtf8(bytes);