From 229bdf6cd88aa36c61305964bc30945831c63c49 Mon Sep 17 00:00:00 2001
From: nishithakbhaskaran <nishithakbhaskaran@gmail.com>
Date: Wed, 20 Nov 2024 14:32:11 +0530
Subject: [PATCH] Fix snakeyaml CVEs

snakeyaml 1.x version causing mend scan CVEs.

Cherry-pick of https://github.com/trinodb/trino/pull/2377

Co-authored-by: Yuya Ebihara <ebyhry@gmail.com>
---
 pom.xml                                       | 23 ++---
 presto-cassandra/pom.xml                      | 14 ++-
 .../cassandra/CassandraQueryRunner.java       | 21 ++---
 ...dedCassandra.java => CassandraServer.java} | 93 +++++++------------
 .../cassandra/CassandraTestingUtils.java      | 26 +++---
 .../cassandra/TestCassandraConnector.java     | 22 +++--
 .../cassandra/TestCassandraDistributed.java   | 13 ++-
 .../TestCassandraIntegrationSmokeTest.java    | 17 ++--
 .../TestCassandraTokenSplitManager.java       | 17 +++-
 .../src/test/resources/cu-cassandra.yaml      | 13 ++-
 presto-elasticsearch/pom.xml                  |  1 -
 11 files changed, 128 insertions(+), 132 deletions(-)
 rename presto-cassandra/src/test/java/com/facebook/presto/cassandra/{EmbeddedCassandra.java => CassandraServer.java} (64%)

diff --git a/pom.xml b/pom.xml
index fde9885f52e01..ea9e9d5668155 100644
--- a/pom.xml
+++ b/pom.xml
@@ -82,6 +82,7 @@
         <dep.avro.version>1.11.4</dep.avro.version>
         <dep.commons.compress.version>1.26.2</dep.commons.compress.version>
         <dep.protobuf-java.version>3.25.5</dep.protobuf-java.version>
+        <dep.snakeyaml.version>2.0</dep.snakeyaml.version>
 
         <!--
           America/Bahia_Banderas has:
@@ -2057,22 +2058,6 @@
                 <version>1.1.3</version>
             </dependency>
 
-            <dependency>
-                <groupId>com.facebook.presto.cassandra</groupId>
-                <artifactId>cassandra-server</artifactId>
-                <version>2.1.16-1</version>
-                <exclusions>
-                    <exclusion>
-                        <groupId>com.googlecode.json-simple</groupId>
-                        <artifactId>json-simple</artifactId>
-                    </exclusion>
-                    <exclusion>
-                        <groupId>io.netty</groupId>
-                        <artifactId>netty-all</artifactId>
-                    </exclusion>
-                </exclusions>
-            </dependency>
-
             <dependency>
                 <groupId>com.facebook.presto</groupId>
                 <artifactId>presto-delta</artifactId>
@@ -2218,6 +2203,12 @@
                 <artifactId>cassandra-driver-core</artifactId>
                 <version>3.11.5</version>
             </dependency>
+
+            <dependency>
+                <groupId>org.yaml</groupId>
+                <artifactId>snakeyaml</artifactId>
+                <version>${dep.snakeyaml.version}</version>
+            </dependency>
         </dependencies>
     </dependencyManagement>
 
diff --git a/presto-cassandra/pom.xml b/presto-cassandra/pom.xml
index 310f32e689d4a..0283700aeaeb2 100644
--- a/presto-cassandra/pom.xml
+++ b/presto-cassandra/pom.xml
@@ -156,8 +156,8 @@
         </dependency>
 
         <dependency>
-            <groupId>com.facebook.presto.cassandra</groupId>
-            <artifactId>cassandra-server</artifactId>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>testcontainers</artifactId>
             <scope>test</scope>
         </dependency>
 
@@ -185,6 +185,16 @@
         </dependency>
     </dependencies>
 
+    <dependencyManagement>
+        <dependencies>
+            <dependency>
+                <groupId>org.slf4j</groupId>
+                <artifactId>slf4j-api</artifactId>
+                <version>1.7.36</version>
+            </dependency>
+        </dependencies>
+    </dependencyManagement>
+
     <build>
         <plugins>
             <plugin>
diff --git a/presto-cassandra/src/test/java/com/facebook/presto/cassandra/CassandraQueryRunner.java b/presto-cassandra/src/test/java/com/facebook/presto/cassandra/CassandraQueryRunner.java
index 8bf56a1727fb6..202f893dbc000 100644
--- a/presto-cassandra/src/test/java/com/facebook/presto/cassandra/CassandraQueryRunner.java
+++ b/presto-cassandra/src/test/java/com/facebook/presto/cassandra/CassandraQueryRunner.java
@@ -34,11 +34,9 @@ private CassandraQueryRunner()
 
     private static boolean tpchLoaded;
 
-    public static synchronized DistributedQueryRunner createCassandraQueryRunner()
+    public static DistributedQueryRunner createCassandraQueryRunner(CassandraServer server)
             throws Exception
     {
-        EmbeddedCassandra.start();
-
         DistributedQueryRunner queryRunner = new DistributedQueryRunner(createCassandraSession("tpch"), 4);
 
         queryRunner.installPlugin(new TpchPlugin());
@@ -46,18 +44,15 @@ public static synchronized DistributedQueryRunner createCassandraQueryRunner()
 
         queryRunner.installPlugin(new CassandraPlugin());
         queryRunner.createCatalog("cassandra", "cassandra", ImmutableMap.of(
-                "cassandra.contact-points", EmbeddedCassandra.getHost(),
-                "cassandra.native-protocol-port", Integer.toString(EmbeddedCassandra.getPort()),
+                "cassandra.contact-points", server.getHost(),
+                "cassandra.native-protocol-port", Integer.toString(server.getPort()),
                 "cassandra.allow-drop-table", "true"));
 
-        if (!tpchLoaded) {
-            createKeyspace(EmbeddedCassandra.getSession(), "tpch");
-            List<TpchTable<?>> tables = TpchTable.getTables();
-            copyTpchTables(queryRunner, "tpch", TINY_SCHEMA_NAME, createCassandraSession("tpch"), tables);
-            for (TpchTable table : tables) {
-                EmbeddedCassandra.refreshSizeEstimates("tpch", table.getTableName());
-            }
-            tpchLoaded = true;
+        createKeyspace(server.getSession(), "tpch");
+        List<TpchTable<?>> tables = TpchTable.getTables();
+        copyTpchTables(queryRunner, "tpch", TINY_SCHEMA_NAME, createCassandraSession("tpch"), tables);
+        for (TpchTable<?> table : tables) {
+            server.refreshSizeEstimates("tpch", table.getTableName());
         }
 
         return queryRunner;
diff --git a/presto-cassandra/src/test/java/com/facebook/presto/cassandra/EmbeddedCassandra.java b/presto-cassandra/src/test/java/com/facebook/presto/cassandra/CassandraServer.java
similarity index 64%
rename from presto-cassandra/src/test/java/com/facebook/presto/cassandra/EmbeddedCassandra.java
rename to presto-cassandra/src/test/java/com/facebook/presto/cassandra/CassandraServer.java
index 9c8f90c271d1e..ca04ae6b116ef 100644
--- a/presto-cassandra/src/test/java/com/facebook/presto/cassandra/EmbeddedCassandra.java
+++ b/presto-cassandra/src/test/java/com/facebook/presto/cassandra/CassandraServer.java
@@ -21,13 +21,11 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.io.Resources;
 import io.airlift.units.Duration;
-import org.apache.cassandra.service.CassandraDaemon;
-
-import javax.management.ObjectName;
+import org.testcontainers.containers.GenericContainer;
 
+import java.io.Closeable;
 import java.io.File;
 import java.io.IOException;
-import java.lang.management.ManagementFactory;
 import java.net.InetSocketAddress;
 import java.nio.file.Files;
 import java.nio.file.Path;
@@ -35,7 +33,6 @@
 import java.util.concurrent.TimeoutException;
 
 import static com.datastax.driver.core.ProtocolVersion.V3;
-import static com.google.common.base.Preconditions.checkState;
 import static com.google.common.io.Files.createTempDir;
 import static com.google.common.io.Files.write;
 import static com.google.common.io.Resources.getResource;
@@ -45,43 +42,38 @@
 import static java.util.concurrent.TimeUnit.MINUTES;
 import static java.util.concurrent.TimeUnit.NANOSECONDS;
 import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.testcontainers.utility.MountableFile.forHostPath;
 import static org.testng.Assert.assertEquals;
 
-public final class EmbeddedCassandra
+public class CassandraServer
+        implements Closeable
+
 {
-    private static Logger log = Logger.get(EmbeddedCassandra.class);
+    private static Logger log = Logger.get(CassandraServer.class);
 
-    private static final String HOST = "127.0.0.1";
     private static final int PORT = 9142;
 
     private static final Duration REFRESH_SIZE_ESTIMATES_TIMEOUT = new Duration(1, MINUTES);
 
-    private static CassandraSession session;
-    private static boolean initialized;
+    private final GenericContainer<?> dockerContainer;
 
-    private EmbeddedCassandra() {}
+    private final CassandraSession session;
 
-    public static synchronized void start()
+    public CassandraServer()
             throws Exception
     {
-        if (initialized) {
-            return;
-        }
-
         log.info("Starting cassandra...");
 
-        System.setProperty("cassandra.config", "file:" + prepareCassandraYaml());
-        System.setProperty("cassandra-foreground", "true");
-        System.setProperty("cassandra.native.epoll.enabled", "false");
-
-        CassandraDaemon cassandraDaemon = new CassandraDaemon();
-        cassandraDaemon.activate();
+        this.dockerContainer = new GenericContainer<>("cassandra:2.1.16")
+                .withExposedPorts(PORT)
+                .withCopyFileToContainer(forHostPath(prepareCassandraYaml()), "/etc/cassandra/cassandra.yaml");
+        this.dockerContainer.start();
 
         Cluster.Builder clusterBuilder = Cluster.builder()
                 .withProtocolVersion(V3)
                 .withClusterName("TestCluster")
                 .addContactPointsWithPorts(ImmutableList.of(
-                        new InetSocketAddress(HOST, PORT)))
+                        new InetSocketAddress(this.dockerContainer.getContainerIpAddress(), this.dockerContainer.getMappedPort(PORT))))
                 .withMaxSchemaAgreementWaitSeconds(30);
 
         ReopeningCluster cluster = new ReopeningCluster(clusterBuilder::build);
@@ -96,12 +88,11 @@ public static synchronized void start()
         }
         catch (RuntimeException e) {
             cluster.close();
-            cassandraDaemon.deactivate();
+            this.dockerContainer.stop();
             throw e;
         }
 
-        EmbeddedCassandra.session = session;
-        initialized = true;
+        this.session = session;
     }
 
     private static String prepareCassandraYaml()
@@ -115,35 +106,25 @@ private static String prepareCassandraYaml()
         Path dataDir = tmpDirPath.resolve("data");
         Files.createDirectory(dataDir);
 
-        String modified = original.replaceAll("\\$\\{data_directory\\}", dataDir.toAbsolutePath().toString());
-
         Path yamlLocation = tmpDirPath.resolve("cu-cassandra.yaml");
-        write(modified, yamlLocation.toFile(), UTF_8);
+        write(original, yamlLocation.toFile(), UTF_8);
 
         return yamlLocation.toAbsolutePath().toString();
     }
 
-    public static synchronized CassandraSession getSession()
+    public CassandraSession getSession()
     {
-        checkIsInitialized();
         return requireNonNull(session, "cluster is null");
     }
 
-    public static synchronized String getHost()
+    public String getHost()
     {
-        checkIsInitialized();
-        return HOST;
+        return dockerContainer.getContainerIpAddress();
     }
 
-    public static synchronized int getPort()
+    public int getPort()
     {
-        checkIsInitialized();
-        return PORT;
-    }
-
-    private static void checkIsInitialized()
-    {
-        checkState(initialized, "EmbeddedCassandra must be started with #start() method before retrieving the cluster retrieval");
+        return dockerContainer.getMappedPort(PORT);
     }
 
     private static void checkConnectivity(CassandraSession session)
@@ -155,7 +136,7 @@ private static void checkConnectivity(CassandraSession session)
         log.info("Cassandra version: %s", version);
     }
 
-    public static void refreshSizeEstimates(String keyspace, String table)
+    public void refreshSizeEstimates(String keyspace, String table)
             throws Exception
     {
         long deadline = System.nanoTime() + REFRESH_SIZE_ESTIMATES_TIMEOUT.roundTo(NANOSECONDS);
@@ -173,27 +154,21 @@ public static void refreshSizeEstimates(String keyspace, String table)
         throw new TimeoutException(format("Attempting to refresh size estimates for table %s.%s has timed out after %s", keyspace, table, REFRESH_SIZE_ESTIMATES_TIMEOUT));
     }
 
-    private static void flushTable(String keyspace, String table)
+    private void flushTable(String keyspace, String table)
             throws Exception
     {
-        ManagementFactory
-                .getPlatformMBeanServer()
-                .invoke(
-                        new ObjectName("org.apache.cassandra.db:type=StorageService"),
-                        "forceKeyspaceFlush",
-                        new Object[] {keyspace, new String[] {table}},
-                        new String[] {"java.lang.String", "[Ljava.lang.String;"});
+        dockerContainer.execInContainer("nodetool", "flush", keyspace, table);
     }
 
-    private static void refreshSizeEstimates()
+    private void refreshSizeEstimates()
             throws Exception
     {
-        ManagementFactory
-                .getPlatformMBeanServer()
-                .invoke(
-                        new ObjectName("org.apache.cassandra.db:type=StorageService"),
-                        "refreshSizeEstimates",
-                        new Object[] {},
-                        new String[] {});
+        dockerContainer.execInContainer("nodetool", "refreshsizeestimates");
+    }
+
+    @Override
+    public void close()
+    {
+        dockerContainer.close();
     }
 }
diff --git a/presto-cassandra/src/test/java/com/facebook/presto/cassandra/CassandraTestingUtils.java b/presto-cassandra/src/test/java/com/facebook/presto/cassandra/CassandraTestingUtils.java
index 791f72c53301b..b8616f7bb03c1 100644
--- a/presto-cassandra/src/test/java/com/facebook/presto/cassandra/CassandraTestingUtils.java
+++ b/presto-cassandra/src/test/java/com/facebook/presto/cassandra/CassandraTestingUtils.java
@@ -75,12 +75,12 @@ public static void createTableClusteringKeys(CassandraSession session, SchemaTab
 
     public static void insertIntoTableClusteringKeys(CassandraSession session, SchemaTableName table, int rowsCount)
     {
-        for (Integer rowNumber = 1; rowNumber <= rowsCount; rowNumber++) {
+        for (int rowNumber = 1; rowNumber <= rowsCount; rowNumber++) {
             Insert insert = QueryBuilder.insertInto(table.getSchemaName(), table.getTableName())
-                    .value("key", "key_" + rowNumber.toString())
+                    .value("key", "key_" + rowNumber)
                     .value("clust_one", "clust_one")
-                    .value("clust_two", "clust_two_" + rowNumber.toString())
-                    .value("clust_three", "clust_three_" + rowNumber.toString());
+                    .value("clust_two", "clust_two_" + rowNumber)
+                    .value("clust_three", "clust_three_" + rowNumber);
             session.execute(insert);
         }
         assertEquals(session.execute("SELECT COUNT(*) FROM " + table).all().get(0).getLong(0), rowsCount);
@@ -103,13 +103,13 @@ public static void createTableMultiPartitionClusteringKeys(CassandraSession sess
 
     public static void insertIntoTableMultiPartitionClusteringKeys(CassandraSession session, SchemaTableName table)
     {
-        for (Integer rowNumber = 1; rowNumber < 10; rowNumber++) {
+        for (int rowNumber = 1; rowNumber < 10; rowNumber++) {
             Insert insert = QueryBuilder.insertInto(table.getSchemaName(), table.getTableName())
-                    .value("partition_one", "partition_one_" + rowNumber.toString())
-                    .value("partition_two", "partition_two_" + rowNumber.toString())
+                    .value("partition_one", "partition_one_" + rowNumber)
+                    .value("partition_two", "partition_two_" + rowNumber)
                     .value("clust_one", "clust_one")
-                    .value("clust_two", "clust_two_" + rowNumber.toString())
-                    .value("clust_three", "clust_three_" + rowNumber.toString());
+                    .value("clust_two", "clust_two_" + rowNumber)
+                    .value("clust_three", "clust_three_" + rowNumber);
             session.execute(insert);
         }
         assertEquals(session.execute("SELECT COUNT(*) FROM " + table).all().get(0).getLong(0), 9);
@@ -131,7 +131,7 @@ public static void createTableClusteringKeysInequality(CassandraSession session,
 
     public static void insertIntoTableClusteringKeysInequality(CassandraSession session, SchemaTableName table, Date date, int rowsCount)
     {
-        for (Integer rowNumber = 1; rowNumber <= rowsCount; rowNumber++) {
+        for (int rowNumber = 1; rowNumber <= rowsCount; rowNumber++) {
             Insert insert = QueryBuilder.insertInto(table.getSchemaName(), table.getTableName())
                     .value("key", "key_1")
                     .value("clust_one", "clust_one")
@@ -222,12 +222,12 @@ public static void createTableAllTypesPartitionKey(CassandraSession session, Sch
 
     private static void insertTestData(CassandraSession session, SchemaTableName table, Date date, int rowsCount)
     {
-        for (Integer rowNumber = 1; rowNumber <= rowsCount; rowNumber++) {
+        for (int rowNumber = 1; rowNumber <= rowsCount; rowNumber++) {
             Insert insert = QueryBuilder.insertInto(table.getSchemaName(), table.getTableName())
-                    .value("key", "key " + rowNumber.toString())
+                    .value("key", "key " + rowNumber)
                     .value("typeuuid", UUID.fromString(String.format("00000000-0000-0000-0000-%012d", rowNumber)))
                     .value("typeinteger", rowNumber)
-                    .value("typelong", rowNumber.longValue() + 1000)
+                    .value("typelong", rowNumber + 1000)
                     .value("typebytes", ByteBuffer.wrap(Ints.toByteArray(rowNumber)).asReadOnlyBuffer())
                     .value("typetimestamp", date)
                     .value("typeansi", "ansi " + rowNumber)
diff --git a/presto-cassandra/src/test/java/com/facebook/presto/cassandra/TestCassandraConnector.java b/presto-cassandra/src/test/java/com/facebook/presto/cassandra/TestCassandraConnector.java
index 64b92c3a21282..3fe4f25c6cca0 100644
--- a/presto-cassandra/src/test/java/com/facebook/presto/cassandra/TestCassandraConnector.java
+++ b/presto-cassandra/src/test/java/com/facebook/presto/cassandra/TestCassandraConnector.java
@@ -42,7 +42,7 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
-import org.testng.annotations.AfterMethod;
+import org.testng.annotations.AfterClass;
 import org.testng.annotations.BeforeClass;
 import org.testng.annotations.Test;
 
@@ -74,7 +74,6 @@
 import static org.testng.Assert.assertTrue;
 import static org.testng.Assert.fail;
 
-@Test(singleThreaded = true)
 public class TestCassandraConnector
 {
     protected static final String INVALID_DATABASE = "totally_invalid_database";
@@ -95,9 +94,11 @@ public class TestCassandraConnector
             Optional.empty(),
             ImmutableMap.of());
     protected String database;
+
     protected SchemaTableName table;
     protected SchemaTableName tableUnpartitioned;
     protected SchemaTableName invalidTable;
+    private CassandraServer server;
     private ConnectorMetadata metadata;
     private ConnectorSplitManager splitManager;
     private ConnectorRecordSetProvider recordSetProvider;
@@ -106,17 +107,17 @@ public class TestCassandraConnector
     public void setup()
             throws Exception
     {
-        EmbeddedCassandra.start();
+        this.server = new CassandraServer();
 
         String keyspace = "test_connector";
-        createTestTables(EmbeddedCassandra.getSession(), keyspace, DATE);
+        createTestTables(server.getSession(), keyspace, DATE);
 
         String connectorId = "cassandra-test";
         CassandraConnectorFactory connectorFactory = new CassandraConnectorFactory(connectorId);
 
         Connector connector = connectorFactory.create(connectorId, ImmutableMap.of(
-                "cassandra.contact-points", EmbeddedCassandra.getHost(),
-                "cassandra.native-protocol-port", Integer.toString(EmbeddedCassandra.getPort())),
+                "cassandra.contact-points", server.getHost(),
+                "cassandra.native-protocol-port", Integer.toString(server.getPort())),
                 new TestingConnectorContext());
 
         metadata = connector.getMetadata(CassandraTransactionHandle.INSTANCE);
@@ -134,14 +135,15 @@ public void setup()
         invalidTable = new SchemaTableName(database, "totally_invalid_table_name");
     }
 
-    @AfterMethod
-    public void tearDown()
+    @Test
+    public void testGetClient()
     {
     }
 
-    @Test
-    public void testGetClient()
+    @AfterClass(alwaysRun = true)
+    public void tearDown()
     {
+        server.close();
     }
 
     @Test
diff --git a/presto-cassandra/src/test/java/com/facebook/presto/cassandra/TestCassandraDistributed.java b/presto-cassandra/src/test/java/com/facebook/presto/cassandra/TestCassandraDistributed.java
index 7307428bb0b50..8b6e01d85d460 100644
--- a/presto-cassandra/src/test/java/com/facebook/presto/cassandra/TestCassandraDistributed.java
+++ b/presto-cassandra/src/test/java/com/facebook/presto/cassandra/TestCassandraDistributed.java
@@ -16,7 +16,7 @@
 import com.facebook.presto.testing.MaterializedResult;
 import com.facebook.presto.testing.QueryRunner;
 import com.facebook.presto.tests.AbstractTestDistributedQueries;
-import org.testng.annotations.Test;
+import org.testng.annotations.AfterClass;
 
 import static com.facebook.presto.common.type.VarcharType.VARCHAR;
 import static com.facebook.presto.testing.MaterializedResult.resultBuilder;
@@ -24,17 +24,24 @@
 
 //Integrations tests fail when parallel, due to a bug or configuration error in the embedded
 //cassandra instance. This problem results in either a hang in Thrift calls or broken sockets.
-@Test(singleThreaded = true)
+
 public class TestCassandraDistributed
         extends AbstractTestDistributedQueries
 {
+    private CassandraServer server;
     @Override
     protected QueryRunner createQueryRunner()
             throws Exception
     {
-        return CassandraQueryRunner.createCassandraQueryRunner();
+        this.server = new CassandraServer();
+        return CassandraQueryRunner.createCassandraQueryRunner(server);
     }
 
+    @AfterClass(alwaysRun = true)
+    public void tearDown()
+    {
+        server.close();
+    }
     @Override
     protected boolean supportsViews()
     {
diff --git a/presto-cassandra/src/test/java/com/facebook/presto/cassandra/TestCassandraIntegrationSmokeTest.java b/presto-cassandra/src/test/java/com/facebook/presto/cassandra/TestCassandraIntegrationSmokeTest.java
index 4639f8af90f8b..508e8d1cc66de 100644
--- a/presto-cassandra/src/test/java/com/facebook/presto/cassandra/TestCassandraIntegrationSmokeTest.java
+++ b/presto-cassandra/src/test/java/com/facebook/presto/cassandra/TestCassandraIntegrationSmokeTest.java
@@ -21,7 +21,7 @@
 import com.facebook.presto.tests.AbstractTestIntegrationSmokeTest;
 import com.google.common.collect.ImmutableList;
 import io.airlift.units.Duration;
-import org.testng.annotations.BeforeClass;
+import org.testng.annotations.AfterClass;
 import org.testng.annotations.Test;
 
 import java.math.BigInteger;
@@ -59,7 +59,6 @@
 import static java.util.stream.Collectors.toList;
 import static org.testng.Assert.assertEquals;
 
-@Test(singleThreaded = true)
 public class TestCassandraIntegrationSmokeTest
         extends AbstractTestIntegrationSmokeTest
 {
@@ -69,13 +68,13 @@ public class TestCassandraIntegrationSmokeTest
     private static final Timestamp DATE_TIME_LOCAL = Timestamp.valueOf(LocalDateTime.of(1970, 1, 1, 3, 4, 5, 0));
     private static final LocalDateTime TIMESTAMP_LOCAL = LocalDateTime.of(1969, 12, 31, 23, 4, 5); // TODO #7122 should match DATE_TIME_LOCAL
 
+    private CassandraServer server;
     private CassandraSession session;
 
-    @BeforeClass
-    public void setUp()
+    @AfterClass(alwaysRun = true)
+    public void tearDown()
     {
-        session = EmbeddedCassandra.getSession();
-        createTestTables(session, KEYSPACE, DATE_TIME_LOCAL);
+        server.close();
     }
 
     @Override
@@ -94,7 +93,11 @@ protected boolean isParameterizedVarcharSupported()
     protected QueryRunner createQueryRunner()
             throws Exception
     {
-        return createCassandraQueryRunner();
+        CassandraServer server = new CassandraServer();
+        this.server = server;
+        this.session = server.getSession();
+        createTestTables(session, KEYSPACE, DATE_TIME_LOCAL);
+        return createCassandraQueryRunner(server);
     }
 
     @Test
diff --git a/presto-cassandra/src/test/java/com/facebook/presto/cassandra/TestCassandraTokenSplitManager.java b/presto-cassandra/src/test/java/com/facebook/presto/cassandra/TestCassandraTokenSplitManager.java
index 74a41c032001a..e18ce7eae042e 100644
--- a/presto-cassandra/src/test/java/com/facebook/presto/cassandra/TestCassandraTokenSplitManager.java
+++ b/presto-cassandra/src/test/java/com/facebook/presto/cassandra/TestCassandraTokenSplitManager.java
@@ -14,6 +14,7 @@
 package com.facebook.presto.cassandra;
 
 import com.facebook.presto.cassandra.CassandraTokenSplitManager.TokenSplit;
+import org.testng.annotations.AfterClass;
 import org.testng.annotations.BeforeClass;
 import org.testng.annotations.Test;
 
@@ -30,6 +31,7 @@ public class TestCassandraTokenSplitManager
     private static final String KEYSPACE = "test_cassandra_token_split_manager_keyspace";
     private static final int PARTITION_COUNT = 1000;
 
+    private CassandraServer server;
     private CassandraSession session;
     private CassandraTokenSplitManager splitManager;
 
@@ -37,19 +39,24 @@ public class TestCassandraTokenSplitManager
     public void setUp()
             throws Exception
     {
-        EmbeddedCassandra.start();
-        session = EmbeddedCassandra.getSession();
+        server = new CassandraServer();
+        session = server.getSession();
         createKeyspace(session, KEYSPACE);
         splitManager = new CassandraTokenSplitManager(session, SPLIT_SIZE, Optional.empty());
     }
 
+    @AfterClass(alwaysRun = true)
+    public void tearDown()
+    {
+        server.close();
+    }
     @Test
     public void testPartitionCountOverride()
             throws Exception
     {
         String tableName = "partition_count_override_table";
         session.execute(format("CREATE TABLE %s.%s (key text PRIMARY KEY)", KEYSPACE, tableName));
-        EmbeddedCassandra.refreshSizeEstimates(KEYSPACE, tableName);
+        server.refreshSizeEstimates(KEYSPACE, tableName);
 
         CassandraTokenSplitManager onlyConfigSplitsPerNode = new CassandraTokenSplitManager(session, SPLIT_SIZE, Optional.of(12_345L));
         assertEquals(12_345L, onlyConfigSplitsPerNode.getTotalPartitionsCount(KEYSPACE, tableName, Optional.empty()));
@@ -70,7 +77,7 @@ public void testEmptyTable()
     {
         String tableName = "empty_table";
         session.execute(format("CREATE TABLE %s.%s (key text PRIMARY KEY)", KEYSPACE, tableName));
-        EmbeddedCassandra.refreshSizeEstimates(KEYSPACE, tableName);
+        server.refreshSizeEstimates(KEYSPACE, tableName);
         List<TokenSplit> splits = splitManager.getSplits(KEYSPACE, tableName, Optional.empty());
         // even for the empty table at least one split must be produced, in case the statistics are inaccurate
         assertEquals(splits.size(), 1);
@@ -86,7 +93,7 @@ public void testNonEmptyTable()
         for (int i = 0; i < PARTITION_COUNT; i++) {
             session.execute(format("INSERT INTO %s.%s (key) VALUES ('%s')", KEYSPACE, tableName, "value" + i));
         }
-        EmbeddedCassandra.refreshSizeEstimates(KEYSPACE, tableName);
+        server.refreshSizeEstimates(KEYSPACE, tableName);
         List<TokenSplit> splits = splitManager.getSplits(KEYSPACE, tableName, Optional.empty());
         assertEquals(splits.size(), PARTITION_COUNT / SPLIT_SIZE);
         session.execute(format("DROP TABLE %s.%s", KEYSPACE, tableName));
diff --git a/presto-cassandra/src/test/resources/cu-cassandra.yaml b/presto-cassandra/src/test/resources/cu-cassandra.yaml
index 181ced7fdb2d3..74bbb7267b4a4 100644
--- a/presto-cassandra/src/test/resources/cu-cassandra.yaml
+++ b/presto-cassandra/src/test/resources/cu-cassandra.yaml
@@ -94,10 +94,10 @@ partitioner: org.apache.cassandra.dht.Murmur3Partitioner
 
 # directories where Cassandra should store data on disk.
 data_file_directories:
-    - ${data_directory}/data
+    - /var/lib/cassandra/data
 
 # commit log
-commitlog_directory: ${data_directory}/commitlog
+commitlog_directory: /var/lib/cassandra/commitlog
 
 # policy for data disk failures:
 # stop: shut down gossip and Thrift, leaving the node effectively dead, but
@@ -160,7 +160,7 @@ row_cache_save_period: 0
 # row_cache_keys_to_save: 100
 
 # saved caches
-saved_caches_directory: ${data_directory}/saved_caches
+saved_caches_directory: /var/lib/cassandra/saved_caches
 
 # commitlog_sync may be either "periodic" or "batch."
 # When in batch mode, Cassandra won't ack writes until the commit log
@@ -280,6 +280,13 @@ rpc_address: localhost
 # port for Thrift to listen for clients on
 rpc_port: 9160
 
+# RPC address to broadcast to drivers and other Cassandra nodes. This cannot
+# be set to 0.0.0.0. If left blank, this will be set to the value of
+# rpc_address. If rpc_address is set to 0.0.0.0, broadcast_rpc_address must
+# be set.
+broadcast_rpc_address: localhost
+
+
 # enable or disable keepalive on rpc connections
 rpc_keepalive: true
 
diff --git a/presto-elasticsearch/pom.xml b/presto-elasticsearch/pom.xml
index d79ea08ab8016..073f3c5518aa2 100644
--- a/presto-elasticsearch/pom.xml
+++ b/presto-elasticsearch/pom.xml
@@ -14,7 +14,6 @@
         <air.main.basedir>${project.parent.basedir}</air.main.basedir>
         <dep.log4j.version>2.17.1</dep.log4j.version>
         <dep.elasticsearch.version>6.0.0</dep.elasticsearch.version>
-        <dep.snakeyaml.version>2.0</dep.snakeyaml.version>
     </properties>
 
     <dependencies>