diff --git a/cassandra-analytics-common/src/main/java/org/apache/cassandra/bridge/CassandraVersion.java b/cassandra-analytics-common/src/main/java/org/apache/cassandra/bridge/CassandraVersion.java
index 167039efd..f524f7bd3 100644
--- a/cassandra-analytics-common/src/main/java/org/apache/cassandra/bridge/CassandraVersion.java
+++ b/cassandra-analytics-common/src/main/java/org/apache/cassandra/bridge/CassandraVersion.java
@@ -20,7 +20,6 @@
 package org.apache.cassandra.bridge;
 
 import java.util.Arrays;
-import java.util.HashSet;
 import java.util.Optional;
 import java.util.Set;
 
@@ -50,7 +49,7 @@ public enum CassandraVersion
         this.number = number;
         this.name = name;
         this.jarBaseName = jarBaseName;
-        this.sstableFormats = new HashSet<>(Arrays.asList(sstableFormats));
+        this.sstableFormats = Set.copyOf(Arrays.asList(sstableFormats));
     }
 
     public int versionNumber()
@@ -68,6 +67,11 @@ public String jarBaseName()
         return jarBaseName;
     }
 
+    public Set<String> supportedSSTableFormats()
+    {
+        return sstableFormats;
+    }
+
     private static final String sstableFormat;
     private static final CassandraVersion[] implementedVersions;
     private static final String[] supportedVersions;
diff --git a/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/testing/TestUtils.java b/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/testing/TestUtils.java
index a233d377f..b78ce4157 100644
--- a/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/testing/TestUtils.java
+++ b/cassandra-analytics-integration-framework/src/main/java/org/apache/cassandra/testing/TestUtils.java
@@ -54,7 +54,7 @@ public final class TestUtils
      * read repairs.
      */
     public static final String CREATE_TEST_TABLE_STATEMENT =
-    "CREATE TABLE IF NOT EXISTS %s (id int, course text, marks int, PRIMARY KEY (id)) WITH read_repair='NONE';";
+    "CREATE TABLE IF NOT EXISTS %s (id int, course text, marks int, PRIMARY KEY (id, course)) WITH read_repair='NONE';";
 
     private TestUtils()
     {
diff --git a/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/CassandraAnalyticsSimpleBtiTest.java b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/CassandraAnalyticsSimpleBtiTest.java
new file mode 100644
index 000000000..0aca727bd
--- /dev/null
+++ b/cassandra-analytics-integration-tests/src/test/java/org/apache/cassandra/analytics/CassandraAnalyticsSimpleBtiTest.java
@@ -0,0 +1,62 @@
+/*
+ * 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.analytics;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.cassandra.bridge.CassandraVersion;
+import org.apache.cassandra.testing.ClusterBuilderConfiguration;
+import org.apache.cassandra.testing.TestUtils;
+
+import static org.assertj.core.api.Assumptions.assumeThat;
+
+
+/**
+ * A simple test that runs a sample read/write Cassandra Analytics job using BTI format SSTable.
+ */
+class CassandraAnalyticsSimpleBtiTest extends CassandraAnalyticsSimpleTest
+{
+    static
+    {
+        System.setProperty("cassandra.analytics.bridges.sstable_format", "bti");
+    }
+
+    @Override
+    protected void beforeClusterProvisioning()
+    {
+        String version = TestUtils.getDTestClusterVersion().getValue();
+        assumeThat(CassandraVersion.fromVersion(version)
+                                   .orElseThrow()
+                                   .supportedSSTableFormats())
+        .as("BTI SSTable format is not supported in %s", version)
+        .contains("bti");
+    }
+
+    @Override
+    protected ClusterBuilderConfiguration testClusterConfiguration()
+    {
+        ClusterBuilderConfiguration conf = super.testClusterConfiguration();
+        Map<String, Object> additionalConf = new HashMap<>(conf.additionalInstanceConfig);
+        additionalConf.put("sstable", Map.of("selected_format", "bti"));
+        conf.additionalInstanceConfig(additionalConf);
+        return conf;
+    }
+}