You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@drill.apache.org by ar...@apache.org on 2018/01/02 14:50:17 UTC
[5/7] drill git commit: DRILL-5973: Support injection of time-bound
pauses in server
DRILL-5973: Support injection of time-bound pauses in server
Support pause injections in the test framework that are time-bound, to allow for testing high latency scenarios.
e.g. delayed server response to the Drill client allows for test a server-induced timeout
Added tests to verify the commit.
closes #1055
Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/3df11e19
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/3df11e19
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/3df11e19
Branch: refs/heads/master
Commit: 3df11e1905ad384a7205d9daa0e2832394a6b890
Parents: 0343518
Author: Kunal Khatua <kk...@maprtech.com>
Authored: Wed Dec 20 12:52:12 2017 -0800
Committer: Arina Ielchiieva <ar...@gmail.com>
Committed: Tue Jan 2 15:43:30 2018 +0200
----------------------------------------------------------------------
.../testing/CountDownLatchInjectionImpl.java | 4 +-
.../drill/exec/testing/ExceptionInjection.java | 2 +-
.../exec/testing/ExecutionControlsInjector.java | 9 +-
.../apache/drill/exec/testing/Injection.java | 8 +-
.../drill/exec/testing/PauseInjection.java | 24 ++++-
.../org/apache/drill/exec/testing/Controls.java | 29 ++++++
.../exec/testing/ControlsInjectionUtil.java | 27 +++++
.../drill/exec/testing/TestPauseInjection.java | 101 +++++++++++++++++++
8 files changed, 194 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/drill/blob/3df11e19/exec/java-exec/src/main/java/org/apache/drill/exec/testing/CountDownLatchInjectionImpl.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/testing/CountDownLatchInjectionImpl.java b/exec/java-exec/src/main/java/org/apache/drill/exec/testing/CountDownLatchInjectionImpl.java
index 7584bad..e4eeeb7 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/testing/CountDownLatchInjectionImpl.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/testing/CountDownLatchInjectionImpl.java
@@ -24,8 +24,6 @@ import com.fasterxml.jackson.annotation.JsonProperty;
import com.google.common.base.Preconditions;
import org.apache.drill.common.concurrent.ExtendedLatch;
-import java.util.concurrent.CountDownLatch;
-
/**
* See {@link org.apache.drill.exec.testing.CountDownLatchInjection} Degenerates to
* {@link org.apache.drill.exec.testing.PauseInjection#pause}, if initialized to zero count. In any case, this injection
@@ -42,7 +40,7 @@ public class CountDownLatchInjectionImpl extends Injection implements CountDownL
@JsonProperty("port") final int port,
@JsonProperty("siteClass") final String siteClass,
@JsonProperty("desc") final String desc) throws InjectionConfigurationException {
- super(address, port, siteClass, desc, 0, 1);
+ super(address, port, siteClass, desc, 0, 1, 0L);
}
@Override
http://git-wip-us.apache.org/repos/asf/drill/blob/3df11e19/exec/java-exec/src/main/java/org/apache/drill/exec/testing/ExceptionInjection.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/testing/ExceptionInjection.java b/exec/java-exec/src/main/java/org/apache/drill/exec/testing/ExceptionInjection.java
index 61f0d67..aff4dba 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/testing/ExceptionInjection.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/testing/ExceptionInjection.java
@@ -44,7 +44,7 @@ public class ExceptionInjection extends Injection {
@JsonProperty("nSkip") final int nSkip,
@JsonProperty("nFire") final int nFire,
@JsonProperty("exceptionClass") String classString) throws InjectionConfigurationException {
- super(address, port, siteClass, desc, nSkip, nFire);
+ super(address, port, siteClass, desc, nSkip, nFire, 0L);
final Class<?> clazz;
try {
clazz = Class.forName(classString);
http://git-wip-us.apache.org/repos/asf/drill/blob/3df11e19/exec/java-exec/src/main/java/org/apache/drill/exec/testing/ExecutionControlsInjector.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/testing/ExecutionControlsInjector.java b/exec/java-exec/src/main/java/org/apache/drill/exec/testing/ExecutionControlsInjector.java
index d8979d2..3a74fee 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/testing/ExecutionControlsInjector.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/testing/ExecutionControlsInjector.java
@@ -17,6 +17,8 @@
*/
package org.apache.drill.exec.testing;
+import java.util.concurrent.TimeUnit;
+
import org.apache.drill.exec.ops.FragmentContext;
import com.google.common.base.Preconditions;
import org.slf4j.Logger;
@@ -81,7 +83,12 @@ public class ExecutionControlsInjector implements ControlsInjector {
executionControls.lookupPauseInjection(this, desc);
if (pauseInjection != null) {
- logger.debug("Pausing at {}", desc);
+ long pauseDuration = pauseInjection.getMsPause();
+ if ( pauseDuration > 0L) {
+ logger.debug("Pausing at {} for {} sec", desc, TimeUnit.MILLISECONDS.toSeconds(pauseDuration) );
+ } else {
+ logger.debug("Pausing at {}", desc);
+ }
pauseInjection.pause();
logger.debug("Resuming at {}", desc);
}
http://git-wip-us.apache.org/repos/asf/drill/blob/3df11e19/exec/java-exec/src/main/java/org/apache/drill/exec/testing/Injection.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/testing/Injection.java b/exec/java-exec/src/main/java/org/apache/drill/exec/testing/Injection.java
index 08ade51..1daf639 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/testing/Injection.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/testing/Injection.java
@@ -32,9 +32,10 @@ public abstract class Injection {
protected final String desc; // description of the injection site; useful for multiple exception injections in a single class
private final AtomicInteger nSkip; // the number of times to skip the injection; starts >= 0
private final AtomicInteger nFire; // the number of times to do the injection, after any skips; starts > 0
+ private final long msPause; // duration of the injection (only applies to pause injections)
protected Injection(final String address, final int port, final String siteClass, final String desc,
- final int nSkip, final int nFire) throws InjectionConfigurationException {
+ final int nSkip, final int nFire, final long msPause) throws InjectionConfigurationException {
if (desc == null || desc.isEmpty()) {
throw new InjectionConfigurationException("Injection desc is null or empty.");
}
@@ -57,6 +58,7 @@ public abstract class Injection {
this.desc = desc;
this.nSkip = new AtomicInteger(nSkip);
this.nFire = new AtomicInteger(nFire);
+ this.msPause = msPause;
}
/**
@@ -81,4 +83,8 @@ public abstract class Injection {
return address == null ||
(address.equals(endpoint.getAddress()) && port == endpoint.getUserPort());
}
+
+ public long getMsPause() {
+ return msPause;
+ }
}
http://git-wip-us.apache.org/repos/asf/drill/blob/3df11e19/exec/java-exec/src/main/java/org/apache/drill/exec/testing/PauseInjection.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/testing/PauseInjection.java b/exec/java-exec/src/main/java/org/apache/drill/exec/testing/PauseInjection.java
index fc4d8ec..89f47c7 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/testing/PauseInjection.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/testing/PauseInjection.java
@@ -21,8 +21,10 @@ import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility;
import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.util.concurrent.TimeUnit;
+
import org.apache.drill.common.concurrent.ExtendedLatch;
-import org.apache.drill.common.exceptions.DrillRuntimeException;
/**
* Injection for a single pause. Pause indefinitely until signalled. This class is used internally for tracking
@@ -43,15 +45,29 @@ public class PauseInjection extends Injection {
@JsonProperty("port") final int port,
@JsonProperty("siteClass") final String siteClass,
@JsonProperty("desc") final String desc,
- @JsonProperty("nSkip") final int nSkip) throws InjectionConfigurationException {
- super(address, port, siteClass, desc, nSkip, 1);
+ @JsonProperty("nSkip") final int nSkip,
+ @JsonProperty("msPause") final long msPause) throws InjectionConfigurationException {
+ //nFire is 1 since we will inject pauses only once
+ super(address, port, siteClass, desc, nSkip, 1, msPause);
}
+ /**
+ * Pause indefinitely, unless a duration exists
+ */
public void pause() {
if (!injectNow()) {
return;
}
- latch.awaitUninterruptibly();
+ if (this.getMsPause() > 0L) {
+ try {
+ latch.await(getMsPause(), TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ //Unpausing self as this is timed
+ unpause();
+ }
+ } else {
+ latch.awaitUninterruptibly();
+ }
}
public void interruptiblePause() throws InterruptedException {
http://git-wip-us.apache.org/repos/asf/drill/blob/3df11e19/exec/java-exec/src/test/java/org/apache/drill/exec/testing/Controls.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/testing/Controls.java b/exec/java-exec/src/test/java/org/apache/drill/exec/testing/Controls.java
index 36ccee3..bb7f737 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/testing/Controls.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/testing/Controls.java
@@ -138,6 +138,20 @@ public class Controls {
}
/**
+ * Adds a time-bound pause injection to the controls builder with the given parameters.
+ *
+ * @param siteClass class where the pause should happen
+ * @param desc descriptor for the pause site in the site class
+ * @param nSkip number of times to skip before firing
+ * @param msPause duration of the pause in millisec
+ * @return this builder
+ */
+ public Builder addTimedPause(final Class<?> siteClass, final String desc, final int nSkip, final long msPause) {
+ injections.add(ControlsInjectionUtil.createTimedPause(siteClass, desc, nSkip, msPause));
+ return this;
+ }
+
+ /**
* Adds a pause injection to the controls builder with the given parameters. The pause is not skipped i.e. the pause
* happens when execution reaches the site.
*
@@ -164,6 +178,21 @@ public class Controls {
}
/**
+ * Adds a time-bound pause injection (for the specified drillbit) to the controls builder with the given parameters.
+ *
+ * @param siteClass class where the pause should happen
+ * @param desc descriptor for the pause site in the site class
+ * @param nSkip number of times to skip before firing
+ * @param msPause duration of the pause in millisec
+ * @return this builder
+ */
+ public Builder addTimedPauseOnBit(final Class<?> siteClass, final String desc,
+ final DrillbitEndpoint endpoint, final int nSkip, final long msPause) {
+ injections.add(ControlsInjectionUtil.createTimedPauseOnBit(siteClass, desc, nSkip, endpoint, msPause));
+ return this;
+ }
+
+ /**
* Adds a pause injection (for the specified drillbit) to the controls builder with the given parameters. The pause
* is not skipped i.e. the pause happens when execution reaches the site.
*
http://git-wip-us.apache.org/repos/asf/drill/blob/3df11e19/exec/java-exec/src/test/java/org/apache/drill/exec/testing/ControlsInjectionUtil.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/testing/ControlsInjectionUtil.java b/exec/java-exec/src/test/java/org/apache/drill/exec/testing/ControlsInjectionUtil.java
index c4b725c..e3b0b0d 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/testing/ControlsInjectionUtil.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/testing/ControlsInjectionUtil.java
@@ -133,6 +133,18 @@ public class ControlsInjectionUtil {
}
/**
+ * Create a time-bound pause injection. Note this format is not directly accepted by the injection mechanism. Use the
+ * {@link Controls} to build exceptions.
+ */
+ public static String createTimedPause(final Class<?> siteClass, final String desc, final int nSkip, final long msPause) {
+ return "{ \"type\" : \"pause\"," +
+ "\"siteClass\" : \"" + siteClass.getName() + "\","
+ + "\"desc\" : \"" + desc + "\","
+ + "\"nSkip\" : " + nSkip + ","
+ + "\"msPause\" : " + msPause + "}";
+ }
+
+ /**
* Create a pause injection on a specific bit. Note this format is not directly accepted by the injection
* mechanism. Use the {@link Controls} to build exceptions.
*/
@@ -147,6 +159,21 @@ public class ControlsInjectionUtil {
}
/**
+ * Create a pause injection on a specific bit. Note this format is not directly accepted by the injection
+ * mechanism. Use the {@link Controls} to build exceptions.
+ */
+ public static String createTimedPauseOnBit(final Class<?> siteClass, final String desc, final int nSkip,
+ final DrillbitEndpoint endpoint, final long msPause) {
+ return "{ \"type\" : \"pause\"," +
+ "\"siteClass\" : \"" + siteClass.getName() + "\","
+ + "\"desc\" : \"" + desc + "\","
+ + "\"nSkip\" : " + nSkip + ","
+ + "\"msPause\" : " + msPause + ","
+ + "\"address\":\"" + endpoint.getAddress() + "\","
+ + "\"port\":\"" + endpoint.getUserPort() + "\"}";
+ }
+
+ /**
* Create a latch injection. Note this format is not directly accepted by the injection mechanism. Use the
* {@link Controls} to build exceptions.
*/
http://git-wip-us.apache.org/repos/asf/drill/blob/3df11e19/exec/java-exec/src/test/java/org/apache/drill/exec/testing/TestPauseInjection.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/testing/TestPauseInjection.java b/exec/java-exec/src/test/java/org/apache/drill/exec/testing/TestPauseInjection.java
index 2e8aa02..571a793 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/testing/TestPauseInjection.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/testing/TestPauseInjection.java
@@ -145,6 +145,34 @@ public class TestPauseInjection extends BaseTestQuery {
}
@Test
+ public void timedPauseInjected() {
+ final long pauseDuration = 2000L;
+ final long expectedDuration = pauseDuration;
+ final ExtendedLatch trigger = new ExtendedLatch(1);
+ final Pointer<Exception> ex = new Pointer<>();
+ final String controls = Controls.newBuilder()
+ .addTimedPause(DummyClass.class, DummyClass.PAUSES, 0, pauseDuration)
+ .build();
+
+ ControlsInjectionUtil.setControls(session, controls);
+
+ final QueryContext queryContext = new QueryContext(session, bits[0].getContext(), QueryId.getDefaultInstance());
+ //We don't need a ResumingThread because this should automatically resume
+
+ // test that the pause happens
+ final DummyClass dummyClass = new DummyClass(queryContext, trigger);
+ final long actualDuration = dummyClass.pauses();
+ assertTrue(String.format("Test should stop for at least %d milliseconds.", expectedDuration),
+ expectedDuration <= actualDuration);
+ assertTrue("No exception should be thrown.", ex.value == null);
+ try {
+ queryContext.close();
+ } catch (final Exception e) {
+ fail("Failed to close query context: " + e);
+ }
+ }
+
+ @Test
public void pauseOnSpecificBit() {
final RemoteServiceSet remoteServiceSet = RemoteServiceSet.getLocalServiceSet();
final ZookeeperHelper zkHelper = new ZookeeperHelper();
@@ -215,4 +243,77 @@ public class TestPauseInjection extends BaseTestQuery {
zkHelper.stopZookeeper();
}
}
+
+
+ @Test
+ public void timedPauseOnSpecificBit() {
+ final RemoteServiceSet remoteServiceSet = RemoteServiceSet.getLocalServiceSet();
+ final ZookeeperHelper zkHelper = new ZookeeperHelper();
+ zkHelper.startZookeeper(1);
+
+ final long pauseDuration = 2000L;
+ final long expectedDuration = pauseDuration;
+
+ try {
+ // Creating two drillbits
+ final Drillbit drillbit1, drillbit2;
+ final DrillConfig drillConfig = zkHelper.getConfig();
+ try {
+ drillbit1 = Drillbit.start(drillConfig, remoteServiceSet);
+ drillbit2 = Drillbit.start(drillConfig, remoteServiceSet);
+ } catch (final DrillbitStartupException e) {
+ throw new RuntimeException("Failed to start two drillbits.", e);
+ }
+
+ final DrillbitContext drillbitContext1 = drillbit1.getContext();
+ final DrillbitContext drillbitContext2 = drillbit2.getContext();
+
+ final UserSession session = UserSession.Builder.newBuilder()
+ .withCredentials(UserCredentials.newBuilder()
+ .setUserName("foo")
+ .build())
+ .withUserProperties(UserProperties.getDefaultInstance())
+ .withOptionManager(drillbitContext1.getOptionManager())
+ .build();
+
+ final DrillbitEndpoint drillbitEndpoint1 = drillbitContext1.getEndpoint();
+ final String controls = Controls.newBuilder()
+ .addTimedPauseOnBit(DummyClass.class, DummyClass.PAUSES, drillbitEndpoint1, 0, pauseDuration)
+ .build();
+
+ ControlsInjectionUtil.setControls(session, controls);
+ {
+ final ExtendedLatch trigger = new ExtendedLatch(1);
+ final Pointer<Exception> ex = new Pointer<>();
+ final QueryContext queryContext = new QueryContext(session, drillbitContext1, QueryId.getDefaultInstance());
+
+ // test that the pause happens
+ final DummyClass dummyClass = new DummyClass(queryContext, trigger);
+ final long actualDuration = dummyClass.pauses();
+ assertTrue(String.format("Test should stop for at least %d milliseconds.", expectedDuration), expectedDuration <= actualDuration);
+ assertTrue("No exception should be thrown.", ex.value == null);
+ try {
+ queryContext.close();
+ } catch (final Exception e) {
+ fail("Failed to close query context: " + e);
+ }
+ }
+
+ {
+ final ExtendedLatch trigger = new ExtendedLatch(1);
+ final QueryContext queryContext = new QueryContext(session, drillbitContext2, QueryId.getDefaultInstance());
+
+ // if the resume did not happen, the test would hang
+ final DummyClass dummyClass = new DummyClass(queryContext, trigger);
+ dummyClass.pauses();
+ try {
+ queryContext.close();
+ } catch (final Exception e) {
+ fail("Failed to close query context: " + e);
+ }
+ }
+ } finally {
+ zkHelper.stopZookeeper();
+ }
+ }
}