diff --git a/build.xml b/build.xml
index 25a47335b9..9932604ac0 100644
--- a/build.xml
+++ b/build.xml
@@ -120,6 +120,7 @@
+
@@ -423,7 +424,7 @@
-
+
@@ -447,6 +448,8 @@
+
+
diff --git a/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java b/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java
index 0016316707..3cb8dacf1e 100644
--- a/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java
+++ b/test/distributed/org/apache/cassandra/distributed/impl/AbstractCluster.java
@@ -30,6 +30,7 @@ import java.util.Set;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -116,6 +117,8 @@ public abstract class AbstractCluster implements ICluster instanceInitializer;
+
private volatile Thread.UncaughtExceptionHandler previousHandler = null;
protected class Wrapper extends DelegatingInvokableInstance implements IUpgradeableInstance
@@ -152,6 +155,8 @@ public abstract class AbstractCluster implements ICluster)Instance::new, classLoader)
.apply(config, classLoader);
}
@@ -261,6 +266,7 @@ public abstract class AbstractCluster implements ICluster();
this.initialVersion = builder.getVersion();
this.filters = new MessageFilters();
+ this.instanceInitializer = builder.getInstanceInitializer();
int generation = GENERATION.incrementAndGet();
for (int i = 0; i < builder.getNodeCount(); ++i)
diff --git a/test/distributed/org/apache/cassandra/distributed/test/ByteBuddyExamples.java b/test/distributed/org/apache/cassandra/distributed/test/ByteBuddyExamples.java
new file mode 100644
index 0000000000..b49572dc4b
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/distributed/test/ByteBuddyExamples.java
@@ -0,0 +1,123 @@
+/*
+ * 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.cassandra.distributed.test;
+
+import java.io.IOException;
+import java.util.concurrent.Callable;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.Test;
+
+import net.bytebuddy.ByteBuddy;
+import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
+import net.bytebuddy.implementation.MethodDelegation;
+import net.bytebuddy.implementation.bind.annotation.SuperCall;
+import org.apache.cassandra.cql3.QueryOptions;
+import org.apache.cassandra.cql3.statements.ModificationStatement;
+import org.apache.cassandra.cql3.statements.SelectStatement;
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.ConsistencyLevel;
+import org.apache.cassandra.service.QueryState;
+import org.apache.cassandra.transport.messages.ResultMessage;
+
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+public class ByteBuddyExamples extends TestBaseImpl
+{
+ @Test
+ public void writeFailureTest() throws Throwable
+ {
+ try(Cluster cluster = init(Cluster.build(1)
+ .withInstanceInitializer(BBFailHelper::install)
+ .start()))
+ {
+ cluster.schemaChange("create table " + KEYSPACE + ".tbl (id int primary key, t int)");
+ try
+ {
+ cluster.coordinator(1).execute("insert into " + KEYSPACE + ".tbl (id, t) values (1, 1)", ConsistencyLevel.ALL);
+ fail("Should fail");
+ }
+ catch (RuntimeException e)
+ {
+ // expected
+ }
+ }
+ }
+
+ public static class BBFailHelper
+ {
+ static void install(ClassLoader cl, int nodeNumber)
+ {
+ new ByteBuddy().redefine(ModificationStatement.class)
+ .method(named("execute"))
+ .intercept(MethodDelegation.to(BBFailHelper.class))
+ .make()
+ .load(cl, ClassLoadingStrategy.Default.INJECTION);
+ }
+ public static ResultMessage execute()
+ {
+ throw new RuntimeException();
+ }
+ }
+
+ @Test
+ public void countTest() throws IOException
+ {
+ try(Cluster cluster = init(Cluster.build(2)
+ .withInstanceInitializer(BBCountHelper::install)
+ .start()))
+ {
+ cluster.schemaChange("create table " + KEYSPACE + ".tbl (id int primary key, t int)");
+ cluster.coordinator(1).execute("select * from " + KEYSPACE + ".tbl;", ConsistencyLevel.ALL);
+ cluster.coordinator(2).execute("select * from " + KEYSPACE + ".tbl;", ConsistencyLevel.ALL);
+ cluster.get(1).runOnInstance(() -> {
+ assertEquals(1, BBCountHelper.count.get());
+ });
+ cluster.get(2).runOnInstance(() -> {
+ assertEquals(0, BBCountHelper.count.get());
+ });
+
+ }
+ }
+
+ public static class BBCountHelper
+ {
+ static AtomicInteger count = new AtomicInteger();
+ static void install(ClassLoader cl, int nodeNumber)
+ {
+ if (nodeNumber != 1)
+ return;
+ new ByteBuddy().rebase(SelectStatement.class)
+ .method(named("execute").and(takesArguments(3)))
+ .intercept(MethodDelegation.to(BBCountHelper.class))
+ .make()
+ .load(cl, ClassLoadingStrategy.Default.INJECTION);
+ }
+
+ public static ResultMessage.Rows execute(QueryState state, QueryOptions options, long queryStartNanoTime, @SuperCall Callable r) throws Exception
+ {
+ count.incrementAndGet();
+ return r.call();
+ }
+ }
+
+}