Fix ClusterDisruptionIT#testAckedIndexing
* Stop threads before logging the list of exceptions
* For the broken case of concurrent iteration in the finally block and the threads not having shut down,
use `CopyOnWriteArrayList` to have concurrency safe iteration
* Closes
@@ -50,6 +50,7 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
@@ -105,7 +106,7 @@ public class ClusterDisruptionIT extends AbstractDisruptionTestCase {
List<Semaphore> semaphores = new ArrayList<>(nodes.size());
final AtomicInteger idGenerator = new AtomicInteger(0);
final AtomicReference<CountDownLatch> countDownLatchRef = new AtomicReference<>();
- final List<Exception> exceptedExceptions = Collections.synchronizedList(new ArrayList<Exception>());
+ final List<Exception> exceptedExceptions = new CopyOnWriteArrayList<>();
logger.info("starting indexers");
try {
@@ -211,6 +212,12 @@ public class ClusterDisruptionIT extends AbstractDisruptionTestCase {
logger.info("done validating (iteration [{}])", iter);
}
} finally {
+ logger.info("shutting down indexers");
+ stop.set(true);
+ for (Thread indexer : indexers) {
+ indexer.interrupt();
+ indexer.join(60000);
+ }
if (exceptedExceptions.size() > 0) {
StringBuilder sb = new StringBuilder();
for (Exception e : exceptedExceptions) {
@@ -218,12 +225,6 @@ public class ClusterDisruptionIT extends AbstractDisruptionTestCase {
}
logger.debug("Indexing exceptions during disruption: {}", sb);
}
- logger.info("shutting down indexers");
- stop.set(true);
- for (Thread indexer : indexers) {
- indexer.interrupt();
- indexer.join(60000);
- }
}
}
