You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by al...@apache.org on 2021/05/23 22:21:26 UTC
[kudu] branch master updated (323d8ea -> d1969e2)
This is an automated email from the ASF dual-hosted git repository.
alexey pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git.
from 323d8ea KUDU-3223: management of per-table limit by kudu CLI
new 6a07524 [java] utility to detect sanitizer type used for kudu CLI
new d1969e2 KUDU-3277 fix concurrent session flush issue in Java client
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../org/apache/kudu/client/AsyncKuduSession.java | 44 +++++--
.../org/apache/kudu/client/TestKuduClient.java | 134 ++++++++++++++++++++-
.../apache/kudu/test/cluster/KuduBinaryInfo.java | 80 ++++++++++++
3 files changed, 250 insertions(+), 8 deletions(-)
[kudu] 01/02: [java] utility to detect sanitizer type used for kudu
CLI
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
alexey pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit 6a0752420fdf46b821038d2f789cc9742a2ac864
Author: Alexey Serbin <al...@apache.org>
AuthorDate: Sat May 22 13:39:53 2021 -0700
[java] utility to detect sanitizer type used for kudu CLI
This patch adds a new test utility to determine the type of the C++
sanitizer type used by the kudu CLI binary, if any. The new utility is
added into the kudu-test-utils package. I found it's useful to find
whether the mini-cluster binaries are built with TSAN/ASAN sanitizers,
so it's possible to skip heavy test scenarios which are not stable when
running against binaries built with ASAN/TSAN support.
I didn't add any tests, but this functionality is going to be used
in a follow-up patch.
Change-Id: I08c978ecda7321038b97c5c885914c02122606c1
Reviewed-on: http://gerrit.cloudera.org:8080/17487
Tested-by: Alexey Serbin <as...@cloudera.com>
Reviewed-by: Grant Henke <gr...@apache.org>
---
.../apache/kudu/test/cluster/KuduBinaryInfo.java | 80 ++++++++++++++++++++++
1 file changed, 80 insertions(+)
diff --git a/java/kudu-test-utils/src/main/java/org/apache/kudu/test/cluster/KuduBinaryInfo.java b/java/kudu-test-utils/src/main/java/org/apache/kudu/test/cluster/KuduBinaryInfo.java
index 8294e33..f65cd31 100644
--- a/java/kudu-test-utils/src/main/java/org/apache/kudu/test/cluster/KuduBinaryInfo.java
+++ b/java/kudu-test-utils/src/main/java/org/apache/kudu/test/cluster/KuduBinaryInfo.java
@@ -17,6 +17,16 @@
package org.apache.kudu.test.cluster;
+import static java.nio.charset.StandardCharsets.UTF_8;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.util.ArrayList;
+import java.util.List;
+
+import com.google.common.base.Joiner;
+import com.google.common.collect.Lists;
import org.apache.yetus.audience.InterfaceAudience;
import org.apache.yetus.audience.InterfaceStability;
@@ -52,4 +62,74 @@ public class KuduBinaryInfo {
public String getSaslDir() {
return saslDir;
}
+
+ /**
+ * The C++ sanitizer type enabled for the kudu CLI binary.
+ */
+ public enum SanitizerType {
+ NONE,
+ ASAN,
+ TSAN,
+ }
+
+ /**
+ * @return sanitizer type for the kudu CLI binary.
+ */
+ public static SanitizerType getSanitizerType() {
+ List<String> vs = getBinaryVersionStrings();
+ if (vs.size() < 1 || !vs.get(0).startsWith("kudu ")) {
+ throw new RuntimeException(String.format(
+ "unexpected version output from kudu binary: %s",
+ Joiner.on("\n").join(vs)));
+ }
+ for (String s : vs) {
+ if (s.equals("ASAN enabled")) {
+ return SanitizerType.ASAN;
+ } else if (s.equals("TSAN enabled")) {
+ return SanitizerType.TSAN;
+ }
+ }
+ return SanitizerType.NONE;
+ }
+
+ /**
+ * @return sequence of strings output by 'kudu --version'
+ */
+ private static List<String> getBinaryVersionStrings() {
+ try {
+ KuduBinaryLocator.ExecutableInfo exeInfo =
+ KuduBinaryLocator.findBinary("kudu");
+ ProcessBuilder pb = new ProcessBuilder(
+ Lists.newArrayList(exeInfo.exePath(), "--version"));
+ pb.environment().putAll(exeInfo.environment());
+ pb.redirectError(ProcessBuilder.Redirect.INHERIT);
+ final Process p = pb.start();
+ List<String> result = new ArrayList<>();
+ try (InputStreamReader isr = new InputStreamReader(p.getInputStream(), UTF_8);
+ BufferedReader br = new BufferedReader(isr)) {
+ while (true) {
+ String line = br.readLine();
+ if (line == null) {
+ break;
+ }
+ result.add(line);
+ }
+ }
+ final int exitCode = p.waitFor();
+ if (exitCode != 0) {
+ // Don't bother reporting the contents of stderr: it should be in the
+ // log of the parent process due to the stderr redirection.
+ throw new RuntimeException(String.format(
+ "unexpected exit code from kudu binary: %d", exitCode));
+ }
+ return result;
+ } catch (IOException e) {
+ throw new RuntimeException(
+ "unexpected exception while trying to run kudu binary", e);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(
+ "unexpected exception while trying to run kudu binary", e);
+ }
+ }
}
[kudu] 02/02: KUDU-3277 fix concurrent session flush issue in Java
client
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
alexey pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit d1969e2623e30bfd62f2463f15e0bf02422d1d84
Author: Alexey Serbin <al...@apache.org>
AuthorDate: Mon May 17 16:11:03 2021 -0700
KUDU-3277 fix concurrent session flush issue in Java client
This patch fixes the issue reported by KUDU-3277. The bug manifested
itself in rare cases when a session running in the AUTO_FLUSH_BACKGROUND
mode was either being flushed explicitly or implicitly (i.e. upon
closing the session) while the AUTO_FLUSH_BACKGROUND session's logic was
flushing its data buffers concurrently as well.
This patch also adds a reproduction scenario for KUDU-3277. The newly
introduced test scenario was reliably failing before the fix:
* 'java.lang.AssertionError: This Deferred was already called'
messages were encountered in the log multiple times with the stack
exactly as described in KUDU-3277
* some flusher threads were unable to join since KuduSession.flush()
would hang (i.e. would not return)
Change-Id: If6aaccc06abf1a2673620ab7c649f51f91999ad9
Reviewed-on: http://gerrit.cloudera.org:8080/17486
Tested-by: Alexey Serbin <as...@cloudera.com>
Reviewed-by: Grant Henke <gr...@apache.org>
---
.../org/apache/kudu/client/AsyncKuduSession.java | 44 +++++--
.../org/apache/kudu/client/TestKuduClient.java | 134 ++++++++++++++++++++-
2 files changed, 170 insertions(+), 8 deletions(-)
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduSession.java b/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduSession.java
index 59eb838..3b0f5c0 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduSession.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduSession.java
@@ -508,8 +508,9 @@ public class AsyncKuduSession implements SessionConfiguration {
* @param buffer the buffer to return to the inactive queue.
*/
private void queueBuffer(Buffer buffer) {
- inactiveBuffers.add(buffer);
- buffer.callbackFlushNotification();
+ if (buffer.callbackFlushNotification()) {
+ inactiveBuffers.add(buffer);
+ }
Deferred<Void> localFlushNotification = flushNotification.getAndSet(new Deferred<>());
localFlushNotification.callback(null);
}
@@ -861,6 +862,7 @@ public class AsyncKuduSession implements SessionConfiguration {
private FlusherTask flusherTask = null;
private Deferred<Void> flushNotification = Deferred.fromResult(null);
+ private boolean flushNotificationFired = false;
public List<BufferedOperation> getOperations() {
return operations;
@@ -884,12 +886,27 @@ public class AsyncKuduSession implements SessionConfiguration {
}
/**
- * Completes the buffer's flush notification. Should be called when the buffer has been
- * successfully flushed.
+ * Completes the buffer's flush notification. Should be called when
+ * the buffer has been successfully flushed.
*/
- void callbackFlushNotification() {
+ boolean callbackFlushNotification() {
LOG.trace("buffer flush notification fired: {}", this);
- flushNotification.callback(null);
+ if (injectLatencyBufferFlushCb) {
+ try {
+ Thread.sleep(randomizer.nextInt(16));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ synchronized (monitor) {
+ if (flushNotificationFired) {
+ // Do nothing: the callback has been called already.
+ return false;
+ }
+ flushNotificationFired = true;
+ flushNotification.callback(null);
+ }
+ return true;
}
/**
@@ -901,6 +918,7 @@ public class AsyncKuduSession implements SessionConfiguration {
LOG.trace("buffer resetUnlocked: {}", this);
operations.clear();
flushNotification = new Deferred<>();
+ flushNotificationFired = false;
flusherTask = null;
}
@@ -976,4 +994,16 @@ public class AsyncKuduSession implements SessionConfiguration {
.toString();
}
}
-}
+
+ private static boolean injectLatencyBufferFlushCb = false;
+
+ /**
+ * Inject latency into {@link Buffer#callbackFlushNotification}.
+ */
+ @InterfaceAudience.LimitedPrivate("Test")
+ static void injectLatencyBufferFlushCb(boolean injectLatency) {
+ injectLatencyBufferFlushCb = injectLatency;
+ LOG.warn("latency injection for Buffer flush notification is {}",
+ injectLatency ? "enabled" : "disabled");
+ }
+}
\ No newline at end of file
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/TestKuduClient.java b/java/kudu-client/src/test/java/org/apache/kudu/client/TestKuduClient.java
index 1d473bb..290ed3b 100644
--- a/java/kudu-client/src/test/java/org/apache/kudu/client/TestKuduClient.java
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/TestKuduClient.java
@@ -39,16 +39,17 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
+import static org.junit.Assume.assumeTrue;
import java.io.Closeable;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.math.BigDecimal;
import java.sql.Date;
-import java.time.LocalDate;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
@@ -56,9 +57,11 @@ import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
import com.google.common.collect.ImmutableList;
import com.stumbleupon.async.Deferred;
@@ -78,6 +81,7 @@ import org.apache.kudu.test.KuduTestHarness.LocationConfig;
import org.apache.kudu.test.KuduTestHarness.MasterServerConfig;
import org.apache.kudu.test.KuduTestHarness.TabletServerConfig;
import org.apache.kudu.test.RandomUtils;
+import org.apache.kudu.test.cluster.KuduBinaryInfo;
import org.apache.kudu.util.DateUtil;
import org.apache.kudu.util.DecimalUtil;
import org.apache.kudu.util.TimestampUtil;
@@ -1419,4 +1423,132 @@ public class TestKuduClient {
OperationResponse respOld = session.apply(insertOld);
assertFalse(respOld.hasRowError());
}
+
+ /**
+ * This is a test scenario to reproduce conditions described in KUDU-3277.
+ * The scenario was failing before the fix:
+ * ** 'java.lang.AssertionError: This Deferred was already called' was
+ * encountered multiple times with the stack exactly as described in
+ * KUDU-3277
+ * ** some flusher threads were unable to join since KuduSession.flush()
+ * would hang (i.e. would not return)
+ */
+ @MasterServerConfig(flags = {
+ // A shorter TTL for tablet locations is necessary to induce more frequent
+ // calls to TabletLookupCB.queueBuffer().
+ "--table_locations_ttl_ms=500",
+ })
+ @Test(timeout = 100000)
+ public void testConcurrentFlush() throws Exception {
+ // This is a very intensive and stressful test scenario, so run it only
+ // against Kudu binaries built without sanitizers.
+ assumeTrue("this scenario is to run against non-sanitized binaries only",
+ KuduBinaryInfo.getSanitizerType() == KuduBinaryInfo.SanitizerType.NONE);
+ try {
+ AsyncKuduSession.injectLatencyBufferFlushCb(true);
+
+ CreateTableOptions opts = new CreateTableOptions()
+ .addHashPartitions(ImmutableList.of("key"), 8)
+ .setRangePartitionColumns(ImmutableList.of("key"));
+
+ Schema schema = ClientTestUtil.getBasicSchema();
+ PartialRow lowerBoundA = schema.newPartialRow();
+ PartialRow upperBoundA = schema.newPartialRow();
+ upperBoundA.addInt("key", 0);
+ opts.addRangePartition(lowerBoundA, upperBoundA);
+
+ PartialRow lowerBoundB = schema.newPartialRow();
+ lowerBoundB.addInt("key", 0);
+ PartialRow upperBoundB = schema.newPartialRow();
+ opts.addRangePartition(lowerBoundB, upperBoundB);
+
+ KuduTable table = client.createTable(TABLE_NAME, schema, opts);
+
+ final CountDownLatch keepRunning = new CountDownLatch(1);
+ final int numSessions = 50;
+
+ List<KuduSession> sessions = new ArrayList<>(numSessions);
+ for (int i = 0; i < numSessions; ++i) {
+ KuduSession session = client.newSession();
+ session.setFlushMode(SessionConfiguration.FlushMode.AUTO_FLUSH_BACKGROUND);
+ sessions.add(session);
+ }
+
+ List<Thread> flushers = new ArrayList<>(numSessions);
+ Random random = RandomUtils.getRandom();
+ {
+ for (int idx = 0; idx < numSessions; ++idx) {
+ final int threadIdx = idx;
+ Thread flusher = new Thread(new Runnable() {
+ @Override
+ public void run() {
+ KuduSession session = sessions.get(threadIdx);
+ try {
+ while (!keepRunning.await(random.nextInt(250), TimeUnit.MILLISECONDS)) {
+ session.flush();
+ assertEquals(0, session.countPendingErrors());
+ }
+ } catch (Exception e) {
+ fail("unexpected exception: " + e);
+ }
+ }
+ });
+ flushers.add(flusher);
+ }
+ }
+
+ final int numRowsPerSession = 10000;
+ final CountDownLatch insertersCompleted = new CountDownLatch(numSessions);
+ List<Thread> inserters = new ArrayList<>(numSessions);
+ {
+ for (int idx = 0; idx < numSessions; ++idx) {
+ final int threadIdx = idx;
+ final int keyStart = threadIdx * numRowsPerSession;
+ Thread inserter = new Thread(new Runnable() {
+ @Override
+ public void run() {
+ KuduSession session = sessions.get(threadIdx);
+ try {
+ for (int key = keyStart; key < keyStart + numRowsPerSession; ++key) {
+ Insert insert = ClientTestUtil.createBasicSchemaInsert(table, key);
+ assertNull(session.apply(insert));
+ }
+ session.flush();
+ } catch (Exception e) {
+ fail("unexpected exception: " + e);
+ }
+ insertersCompleted.countDown();
+ }
+ });
+ inserters.add(inserter);
+ }
+ }
+
+ for (Thread flusher : flushers) {
+ flusher.start();
+ }
+ for (Thread inserter : inserters) {
+ inserter.start();
+ }
+
+ // Wait for the inserter threads to finish.
+ insertersCompleted.await();
+ // Signal the flusher threads to stop.
+ keepRunning.countDown();
+
+ for (Thread inserter : inserters) {
+ inserter.join();
+ }
+ for (Thread flusher : flushers) {
+ flusher.join();
+ }
+
+ KuduScanner scanner = new KuduScanner.KuduScannerBuilder(asyncClient, table)
+ .readMode(AsyncKuduScanner.ReadMode.READ_YOUR_WRITES)
+ .build();
+ assertEquals(numSessions * numRowsPerSession, countRowsInScan(scanner));
+ } finally {
+ AsyncKuduSession.injectLatencyBufferFlushCb(false);
+ }
+ }
}