diff --git a/README.md b/README.md
new file mode 100644
index 0000000..7b48652
--- /dev/null
+++ b/README.md
@@ -0,0 +1,86 @@
+# Cassandra diff
+
+## Configuration
+See `spark-job/localconfig.yaml` for an example config.
+
+## Custom cluster providers
+To make it easy to run in any environment the cluster providers are pluggable - there are two interfaces to implement.
+First, the `ClusterProvider` interface is used to create a connection to the clusters, and it is configured using
+`JobConfiguration#clusterConfig` (see below).
+### cluster_config
+This section has 3 parts - `source`, `target` and `metadata` where source and target describes the clusters that should
+be compared and metadata describes where we store information about any mismatches and the progress the job has done. 
+Metadata can be stored in one of the source/target clusters or in a separate cluster.
+
+The fields under source/target/metadata are passed in to the `ClusterProvider` (described by `impl`) as a map, so any
+custom cluster providers can be configured here.
+
+## Setting up clusters for diff
+One way of setting up clusters for diff is to restore a snapshot to two different clusters and then modifying one 
+of the clusters to be able to make sure that the queries still return the same results. This could include 
+upgrades/replacements/bounces/decommission/expansion. 
+
+## Environment variables
+Currently usernames and passwords are set as environment variables when running the diff tool and the api server:
+
+* `diff.cluster.<identifier>.cql_user` - the user name to use
+* `diff.cluster.<identifier>.cql_password` - password
+
+where `<identifier>` should be `source`, `target` and `metadata` for the username/password combinations for the
+matching clusters in the configuration.
+
+## Example
+This example starts two cassandra single-node clusters in docker, runs stress to populate them and then runs diff 
+to make sure the data matches;
+
+You need to have docker and spark setup.
+
+```shell script
+$ git clone <wherever>/cassandra-diff.git
+$ cd cassandra-diff
+$ mvn package
+$ docker run --name cas-src -d  -p 9042:9042 cassandra:3.0.18
+$ docker run --name cas-tgt -d  -p 9043:9042 cassandra:latest
+$ docker exec cas-src cassandra-stress write n=1k
+$ docker exec cas-tgt cassandra-stress write n=1k
+$ spark-submit --verbose --files ./spark-job/localconfig.yaml --class org.apache.cassandra.diff.DiffJob spark-job/target/spark-job-0.1-SNAPSHOT.jar localconfig.yaml
+# ... logs
+INFO  DiffJob:124 - FINISHED: {standard1=Matched Partitions - 1000, Mismatched Partitions - 0, Partition Errors - 0, Partitions Only In Source - 0, Partitions Only In Target - 0, Skipped Partitions - 0, Matched Rows - 1000, Matched Values - 6000, Mismatched Values - 0 }
+## start api-server:
+$ mvn install
+$ cd api-server
+$ mvn exec:java
+$ curl -s localhost:8089/jobs/recent | python -mjson.tool
+  [
+      {
+          "jobId": "99b8d556-07ed-4bfd-b978-7d9b7b2cc21a",
+          "buckets": 100,
+          "keyspace": "keyspace1",
+          "tables": [
+              "standard1"
+          ],
+          "sourceClusterName": "local_test_1",
+          "sourceClusterDesc": "ContactPoints Cluster: name=name, dc=datacenter1, contact points= [127.0.0.1]",
+          "targetClusterName": "local_test_2",
+          "targetClusterDesc": "ContactPoints Cluster: name=name, dc=datacenter1, contact points= [127.0.0.1]",
+          "tasks": 10000,
+          "start": "2019-08-16T11:47:36.123Z"
+      }
+  ]
+$ curl -s localhost:8089/jobs/99b8d556-07ed-4bfd-b978-7d9b7b2cc21a/results | python -mjson.tool
+  [
+      {
+          "jobId": "99b8d556-07ed-4bfd-b978-7d9b7b2cc21a",
+          "table": "standard1",
+          "matchedPartitions": 1000,
+          "mismatchedPartitions": 0,
+          "matchedRows": 1000,
+          "matchedValues": 6000,
+          "mismatchedValues": 0,
+          "onlyInSource": 0,
+          "onlyInTarget": 0,
+          "skippedPartitions": 0
+      }
+  ]
+
+```
diff --git a/api-server/README.md b/api-server/README.md
new file mode 100644
index 0000000..254601c
--- /dev/null
+++ b/api-server/README.md
@@ -0,0 +1,117 @@
+# Diff API Server
+## Configuration
+See main project README - the api server reads the `metadata_options` and `cluster_config.metadata` to connect
+
+## Running locally
+`mvn exec:java`
+
+## Endpoints
+### `/jobs/running/id`
+Returns the ids of currently running jobs
+
+### `/jobs/running`
+Summaries of all currently running jobs
+
+### `/jobs/recent`
+Summaries of recently run jobs
+Example:
+```shell script
+$ curl -s localhost:8089/jobs/recent | python -mjson.tool
+  [
+      {
+          "jobId": "99b8d556-07ed-4bfd-b978-7d9b7b2cc21a",
+          "buckets": 100,
+          "keyspace": "keyspace1",
+          "tables": [
+              "standard1"
+          ],
+          "sourceClusterName": "local_test_1",
+          "sourceClusterDesc": "ContactPoints Cluster: name=name, dc=datacenter1, contact points= [127.0.0.1]",
+          "targetClusterName": "local_test_2",
+          "targetClusterDesc": "ContactPoints Cluster: name=name, dc=datacenter1, contact points= [127.0.0.1]",
+          "tasks": 10000,
+          "start": "2019-08-16T11:47:36.123Z"
+      }
+  ]
+
+```
+
+### `/jobs/{jobid}`
+Summary about a single job
+Example:
+```shell script
+$ curl -s localhost:8089/jobs/99b8d556-07ed-4bfd-b978-7d9b7b2cc21a | python -mjson.tool
+  {
+      "jobId": "99b8d556-07ed-4bfd-b978-7d9b7b2cc21a",
+      "buckets": 100,
+      "keyspace": "keyspace1",
+      "tables": [
+          "standard1"
+      ],
+      "sourceClusterName": "local_test_1",
+      "sourceClusterDesc": "ContactPoints Cluster: name=name, dc=datacenter1, contact points= [127.0.0.1]",
+      "targetClusterName": "local_test_2",
+      "targetClusterDesc": "ContactPoints Cluster: name=name, dc=datacenter1, contact points= [127.0.0.1]",
+      "tasks": 10000,
+      "start": "2019-08-16T11:47:36.123Z"
+  }
+```
+
+### `/jobs/{jobid}/results`
+The results for the given job.
+Example:
+```shell script
+$ curl -s localhost:8089/jobs/99b8d556-07ed-4bfd-b978-7d9b7b2cc21a/results | python -mjson.tool
+[
+    {
+        "jobId": "99b8d556-07ed-4bfd-b978-7d9b7b2cc21a",
+        "table": "standard1",
+        "matchedPartitions": 1000,
+        "mismatchedPartitions": 0,
+        "matchedRows": 1000,
+        "matchedValues": 6000,
+        "mismatchedValues": 0,
+        "onlyInSource": 0,
+        "onlyInTarget": 0,
+        "skippedPartitions": 0
+    }
+]
+```
+
+### `/jobs/{jobid}/status`
+Current status for a job, shows how many splits have been finished.
+Example:
+```shell script
+$ curl -s localhost:8089/jobs/99b8d556-07ed-4bfd-b978-7d9b7b2cc21a/status | python -mjson.tool
+{
+    "jobId": "99b8d556-07ed-4bfd-b978-7d9b7b2cc21a",
+    "completedByTable": {
+        "standard1": 10000
+    }
+}
+```
+
+### `/jobs/{jobid}/mismatches`
+Number of mismatches for the job.
+```shell script
+$  curl -s localhost:8089/jobs/99b8d556-07ed-4bfd-b978-7d9b7b2cc21a/mismatches | python -mjson.tool
+  {
+      "jobId": "99b8d556-07ed-4bfd-b978-7d9b7b2cc21a",
+      "mismatchesByTable": {}
+  }
+```
+
+### `/jobs/{jobid}/errors/summary`
+Summary of the number errors for the job. 
+
+### `/jobs/{jobid}/errors/ranges`
+Lists failed ranges for the job.
+
+### `/jobs/{jobid}/errors`
+Details about the job errors.
+
+### `/jobs/by-start-date/{started-after}`
+### `/jobs/by-start-date/{started-after}/{started-before}`
+### `/jobs/by-source-cluster/{source}`
+### `/jobs/by-target-cluster/{target}`
+### `/jobs/by-keyspace/{keyspace}`
diff --git a/api-server/pom.xml b/api-server/pom.xml
new file mode 100644
index 0000000..ab431da
--- /dev/null
+++ b/api-server/pom.xml
@@ -0,0 +1,119 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ 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.
+  -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
+		 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+      <groupId>org.apache.cassandra.diff</groupId>
+      <artifactId>diff</artifactId>
+      <version>0.1-SNAPSHOT</version>
+    </parent>
+
+    <modelVersion>4.0.0</modelVersion>
+    <artifactId>api-server</artifactId>
+    <packaging>jar</packaging>
+
+    <dependencies>
+
+      <!-- compile dependencies -->
+      <dependency>
+        <groupId>org.apache.cassandra.diff</groupId>
+        <artifactId>common</artifactId>
+        <version>${project.parent.version}</version>
+      </dependency>
+
+      <dependency>
+        <groupId>org.apache.cxf</groupId>
+        <artifactId>cxf-rt-frontend-jaxrs</artifactId>
+        <version>3.1.7</version>
+      </dependency>
+
+      <dependency>
+        <groupId>org.apache.cxf</groupId>
+        <artifactId>cxf-rt-transports-http-jetty</artifactId>
+        <version>3.1.7</version>
+      </dependency>
+
+      <dependency>
+        <groupId>joda-time</groupId>
+        <artifactId>joda-time</artifactId>
+        <version>2.9.9</version>
+      </dependency>
+
+      <dependency>
+        <groupId>com.fasterxml.jackson.core</groupId>
+        <artifactId>jackson-databind</artifactId>
+        <version>2.9.9.2</version>
+      </dependency>
+
+      <!-- provided dependencies -->
+      <dependency>
+        <groupId>org.jetbrains</groupId>
+        <artifactId>annotations</artifactId>
+      </dependency>
+
+      <!-- runtime dependencies -->
+      <dependency>
+        <groupId>org.apache.logging.log4j</groupId>
+        <artifactId>log4j-slf4j-impl</artifactId>
+      </dependency>
+
+      <!-- test dependencies -->
+      <dependency>
+        <groupId>junit</groupId>
+        <artifactId>junit</artifactId>
+      </dependency>
+
+      <dependency>
+        <groupId>org.assertj</groupId>
+        <artifactId>assertj-core</artifactId>
+        <version>3.4.1</version>
+        <scope>test</scope>
+      </dependency>
+
+    </dependencies>
+  <build>
+    <plugins>
+      <plugin>
+        <groupId>org.codehaus.mojo</groupId>
+        <artifactId>exec-maven-plugin</artifactId>
+        <version>1.6.0</version>
+        <executions>
+          <execution>
+            <goals>
+              <goal>java</goal>
+            </goals>
+          </execution>
+        </executions>
+        <configuration>
+          <mainClass>org.apache.cassandra.diff.api.DiffAPIServer</mainClass>
+          <arguments>
+            <argument>
+              ../spark-job/localconfig.yaml
+            </argument>
+          </arguments>
+        </configuration>
+      </plugin>
+    </plugins>
+  </build>
+
+</project>
diff --git a/api-server/src/main/java/org/apache/cassandra/diff/api/DiffAPIServer.java b/api-server/src/main/java/org/apache/cassandra/diff/api/DiffAPIServer.java
new file mode 100644
index 0000000..6f14128
--- /dev/null
+++ b/api-server/src/main/java/org/apache/cassandra/diff/api/DiffAPIServer.java
@@ -0,0 +1,57 @@
+/*
+ * 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.diff.api;
+
+import java.io.IOException;
+
+import com.google.common.collect.Lists;
+
+import org.apache.cassandra.diff.YamlJobConfiguration;
+import org.apache.cassandra.diff.api.resources.DiffJobsResource;
+import org.apache.cassandra.diff.api.resources.HealthResource;
+import org.apache.cxf.endpoint.Server;
+import org.apache.cxf.jaxrs.JAXRSServerFactoryBean;
+import org.apache.cxf.jaxrs.lifecycle.SingletonResourceProvider;
+
+public class DiffAPIServer {
+    public static void main(String[] args) throws IOException {
+        String filename = args[0];
+        JAXRSServerFactoryBean factoryBean = new JAXRSServerFactoryBean();
+
+        DiffJobsResource diffResource = new DiffJobsResource(YamlJobConfiguration.load(filename));
+        factoryBean.setResourceProviders(Lists.newArrayList(new SingletonResourceProvider(diffResource),
+                                                            new SingletonResourceProvider(new HealthResource())));
+        factoryBean.setAddress("http://localhost:8089/");
+        Server server = factoryBean.create();
+
+        try {
+            server.start();
+            System.in.read();
+        } catch (Throwable t) {
+            t.printStackTrace(System.out);
+            throw t;
+        } finally {
+            diffResource.close();
+            if (server.isStarted())
+                server.stop();
+            System.exit(0);
+        }
+    }
+}
diff --git a/api-server/src/main/java/org/apache/cassandra/diff/api/resources/DiffJobsResource.java b/api-server/src/main/java/org/apache/cassandra/diff/api/resources/DiffJobsResource.java
new file mode 100644
index 0000000..564f960
--- /dev/null
+++ b/api-server/src/main/java/org/apache/cassandra/diff/api/resources/DiffJobsResource.java
@@ -0,0 +1,228 @@
+/*
+ * 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.diff.api.resources;
+
+import java.util.SortedSet;
+import java.util.UUID;
+import javax.ws.rs.*;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+
+import com.google.common.collect.Sets;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.diff.JobConfiguration;
+import org.apache.cassandra.diff.api.services.DBService;
+import com.fasterxml.jackson.annotation.JsonAutoDetect;
+import com.fasterxml.jackson.annotation.PropertyAccessor;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.joda.time.DateTime;
+import org.joda.time.DateTimeZone;
+import org.joda.time.format.DateTimeFormat;
+import org.joda.time.format.DateTimeFormatter;
+
+import static com.google.common.base.Strings.isNullOrEmpty;
+
+@Path("/jobs")
+public class DiffJobsResource {
+
+    private static final Logger logger = LoggerFactory.getLogger(DiffJobsResource.class);
+    private static final String DATE_FORMAT = "yyyy-MM-dd";
+    private static final DateTimeFormatter DATE_FORMATTER = DateTimeFormat.forPattern(DATE_FORMAT).withZoneUTC();
+    private static final ObjectMapper MAPPER = new ObjectMapper().setVisibility(PropertyAccessor.FIELD,
+                                                                                JsonAutoDetect.Visibility.NON_PRIVATE);
+    private final DBService dbService;
+
+    public DiffJobsResource(JobConfiguration conf) {
+        dbService = new DBService(conf);
+    }
+
+    @GET
+    @Path("/running/id")
+    @Produces(MediaType.APPLICATION_JSON)
+    public Response getRunningJobIds() {
+        return response(dbService.fetchRunningJobs());
+    }
+
+    @GET
+    @Path("/running")
+    @Produces(MediaType.APPLICATION_JSON)
+    public Response getRunningJobs() {
+        return response(dbService.fetchJobSummaries(dbService.fetchRunningJobs()));
+    }
+
+    @GET
+    @Path("/recent")
+    @Produces(MediaType.APPLICATION_JSON)
+    public Response getRecentJobs() {
+
+        SortedSet<DBService.JobSummary> recentJobs = Sets.newTreeSet(DBService.JobSummary.COMPARATOR.reversed());
+        DateTime now = DateTime.now(DateTimeZone.UTC);
+        DateTime maxStartDate = now;
+        DateTime minStartDate = maxStartDate.minusDays(30);
+        recentJobs.addAll(dbService.fetchJobsStartedBetween(minStartDate, maxStartDate));
+
+        while (recentJobs.size() < 10 && (maxStartDate.compareTo(now.minusDays(90)) >= 0)) {
+            maxStartDate = minStartDate;
+            minStartDate = minStartDate.minusDays(30);
+            recentJobs.addAll(dbService.fetchJobsStartedBetween(minStartDate, maxStartDate));
+        }
+
+        return response(recentJobs);
+    }
+
+    @GET
+    @Path("/{jobid}")
+    @Produces(MediaType.APPLICATION_JSON)
+    public Response getJob(@PathParam("jobid") final String jobId) {
+        return response(dbService.fetchJobSummary(UUID.fromString(jobId)));
+    }
+
+    @GET
+    @Path("/{jobid}/results")
+    @Produces(MediaType.APPLICATION_JSON)
+    public Response getJobResults(@PathParam("jobid") final String jobId) {
+        return response(dbService.fetchJobResults(UUID.fromString(jobId)));
+    }
+
+    @GET
+    @Path("/{jobid}/status")
+    @Produces(MediaType.APPLICATION_JSON)
+    public Response getJobStatus(@PathParam("jobid") final String jobId) {
+        return response(dbService.fetchJobStatus(UUID.fromString(jobId)));
+    }
+
+    @GET
+    @Path("/{jobid}/mismatches")
+    @Produces(MediaType.APPLICATION_JSON)
+    public Response getJobMismatches(@PathParam("jobid") final String jobId) {
+        return response(dbService.fetchMismatches(UUID.fromString(jobId)));
+    }
+
+    @GET
+    @Path("/{jobid}/errors/summary")
+    @Produces(MediaType.APPLICATION_JSON)
+    public Response getJobErrorSummary(@PathParam("jobid") final String jobId) {
+        return response(dbService.fetchErrorSummary(UUID.fromString(jobId)));
+    }
+
+    @GET
+    @Path("/{jobid}/errors/ranges")
+    @Produces(MediaType.APPLICATION_JSON)
+    public Response getJobErrorRanges(@PathParam("jobid") final String jobId) {
+        return response(dbService.fetchErrorRanges(UUID.fromString(jobId)));
+    }
+
+    @GET
+    @Path("/{jobid}/errors")
+    @Produces(MediaType.APPLICATION_JSON)
+    public Response getJobErrorDetail(@PathParam("jobid") final String jobId) {
+        return response(dbService.fetchErrorDetail(UUID.fromString(jobId)));
+    }
+
+    @GET
+    @Path("/by-start-date/{started-after}")
+    @Produces(MediaType.APPLICATION_JSON)
+    public Response getJobsStartedSince(@PathParam("started-after") final String minStart) {
+
+        DateTime minStartDate = parseDate(minStart);
+        if (minStartDate == null)
+            return Response.status(400).entity("Invalid date, please supply in the format yyyy-MM-dd").build();
+
+        DateTime maxStartDate = DateTime.now(DateTimeZone.UTC);
+
+        return getJobsStartedBetween(minStartDate, maxStartDate);
+    }
+
+    @GET
+    @Path("/by-start-date/{started-after}/{started-before}")
+    @Produces(MediaType.APPLICATION_JSON)
+    public Response getJobsStartedSince(@PathParam("started-after") final String minStart,
+                                        @PathParam("started-before") final String maxStart) {
+
+        DateTime minStartDate = parseDate(minStart);
+        if (minStartDate == null)
+            return Response.status(400).entity("Invalid date, please supply in the format yyyy-MM-dd").build();
+
+        DateTime maxStartDate = null;
+        if (isNullOrEmpty(maxStart)) {
+            DateTime.now(DateTimeZone.UTC);
+        }
+        else {
+            maxStartDate = parseDate(maxStart);
+            if (maxStartDate == null)
+                return Response.status(400).entity("Invalid date, please supply in the format yyyy-MM-dd").build();
+            if (maxStartDate.compareTo(minStartDate) < 0)
+                return Response.status(400).entity("Invalid date range, started-before cannot be earlier than started-after").build();
+        }
+
+        return getJobsStartedBetween(minStartDate, maxStartDate);
+    }
+
+    @GET
+    @Path("/by-source-cluster/{source}")
+    @Produces(MediaType.APPLICATION_JSON)
+    public Response getJobsBySourceCluster(@PathParam("source") final String sourceClusterName) {
+        return response(dbService.fetchJobsForSourceCluster(sourceClusterName));
+    }
+
+    @GET
+    @Path("/by-target-cluster/{target}")
+    @Produces(MediaType.APPLICATION_JSON)
+    public Response getJobsByTargetCluster(@PathParam("target") final String targetClusterName) {
+        return response(dbService.fetchJobsForTargetCluster(targetClusterName));
+    }
+
+    @GET
+    @Path("/by-keyspace/{keyspace}")
+    @Produces(MediaType.APPLICATION_JSON)
+    public Response getJobsByKeyspace(@PathParam("keyspace") final String keyspace) {
+        return response(dbService.fetchJobsForKeyspace(keyspace));
+    }
+
+    private Response response(Object data) {
+        try {
+            return Response.ok().entity(MAPPER.writer().writeValueAsString(data)).build();
+        }
+        catch (JsonProcessingException e) {
+            logger.error("JSON Processing error", e);
+            return Response.serverError().entity("Error constructing JSON response").build();
+        }
+    }
+
+    private Response getJobsStartedBetween(DateTime minStartDate, DateTime maxStartDate) {
+        return response(dbService.fetchJobsStartedBetween(minStartDate, maxStartDate));
+    }
+
+    private DateTime parseDate(String s) {
+        try {
+            return DATE_FORMATTER.parseDateTime(s);
+        } catch (IllegalArgumentException e) {
+            return null;
+        }
+    }
+
+    public void close() {
+        dbService.close();
+    }
+}
diff --git a/api-server/src/main/java/org/apache/cassandra/diff/api/resources/HealthResource.java b/api-server/src/main/java/org/apache/cassandra/diff/api/resources/HealthResource.java
new file mode 100644
index 0000000..d101615
--- /dev/null
+++ b/api-server/src/main/java/org/apache/cassandra/diff/api/resources/HealthResource.java
@@ -0,0 +1,75 @@
+/*
+ * 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.diff.api.resources;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.ws.rs.GET;
+import javax.ws.rs.Path;
+import javax.ws.rs.Produces;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+
+/**
+ * Health Check: not terribly useful yet.
+ * TODO: add DB connection check
+ */
+@Path("/__health")
+public class HealthResource {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(HealthResource.class);
+
+  private static Boolean overrideHealthStatus = null;
+
+  public HealthResource() {
+    LOGGER.debug("New HealthResource created.");
+  }
+
+  public static Boolean overrideHealthStatusTo(Boolean newStatus) {
+    Boolean oldStatus = overrideHealthStatus;
+    overrideHealthStatus = newStatus;
+    LOGGER.debug("Changed overrideHealthStatus from {} to {}", oldStatus, overrideHealthStatus);
+    return oldStatus;
+  }
+
+  @GET
+  @Produces(MediaType.APPLICATION_JSON)
+  public Response health() {
+    if (isHealthy()) {
+      return Response
+          .ok("All is good.")
+          .build();
+    } else {
+      return Response
+          .serverError()
+          .entity("Not Healthy.")
+          .build();
+    }
+  }
+
+  private boolean isHealthy() {
+    if (overrideHealthStatus == null) {
+      return true;
+    }
+    return overrideHealthStatus;
+  }
+
+}
diff --git a/api-server/src/main/java/org/apache/cassandra/diff/api/services/DBService.java b/api-server/src/main/java/org/apache/cassandra/diff/api/services/DBService.java
new file mode 100644
index 0000000..1bf1f88
--- /dev/null
+++ b/api-server/src/main/java/org/apache/cassandra/diff/api/services/DBService.java
@@ -0,0 +1,546 @@
+/*
+ * 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.diff.api.services;
+
+import java.io.Closeable;
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.datastax.driver.core.*;
+import com.datastax.driver.core.utils.UUIDs;
+import org.apache.cassandra.diff.ClusterProvider;
+import org.apache.cassandra.diff.JobConfiguration;
+import org.jetbrains.annotations.NotNull;
+import org.joda.time.DateTime;
+import org.joda.time.DateTimeZone;
+import org.joda.time.Days;
+
+// TODO cache jobsummary
+// TODO fix exception handling
+public class DBService implements Closeable {
+    private static final Logger logger = LoggerFactory.getLogger(DBService.class);
+    private static final long QUERY_TIMEOUT_MS = 3000;
+
+    private final Cluster cluster;
+    private final Session session;
+    private final String diffKeyspace;
+
+    private final PreparedStatement runningJobsStatement;
+    private final PreparedStatement jobSummaryStatement;
+    private final PreparedStatement jobResultStatement;
+    private final PreparedStatement jobStatusStatement;
+    private final PreparedStatement jobMismatchesStatement;
+    private final PreparedStatement jobErrorSummaryStatement;
+    private final PreparedStatement jobErrorRangesStatement;
+    private final PreparedStatement jobErrorDetailStatement;
+    private final PreparedStatement jobsStartDateStatement;
+    private final PreparedStatement jobsForSourceStatement;
+    private final PreparedStatement jobsForTargetStatement;
+    private final PreparedStatement jobsForKeyspaceStatement;
+
+    public DBService(JobConfiguration config) {
+        logger.info("Initializing DBService");
+        ClusterProvider provider = ClusterProvider.getProvider(config.clusterConfig("metadata"), "metadata");
+        cluster = provider.getCluster();
+
+        session = cluster.connect();
+        diffKeyspace = config.metadataOptions().keyspace;
+        runningJobsStatement = session.prepare(String.format(
+            " SELECT job_id " +
+            " FROM %s.running_jobs", diffKeyspace));
+        jobSummaryStatement = session.prepare(String.format(
+            " SELECT " +
+            "   job_id," +
+            "   job_start_time," +
+            "   buckets," +
+            "   keyspace_name, " +
+            "   table_names, " +
+            "   source_cluster_name," +
+            "   source_cluster_desc," +
+            "   target_cluster_name," +
+            "   target_cluster_desc," +
+            "   total_tasks " +
+            " FROM %s.job_summary" +
+            " WHERE job_id = ?", diffKeyspace));
+        jobResultStatement = session.prepare(String.format(
+            " SELECT " +
+            "   job_id," +
+            "   table_name, " +
+            "   matched_partitions," +
+            "   mismatched_partitions,"   +
+            "   matched_rows,"   +
+            "   matched_values,"   +
+            "   mismatched_values,"   +
+            "   partitions_only_in_source,"   +
+            "   partitions_only_in_target,"   +
+            "   skipped_partitions"   +
+            " FROM %s.job_results" +
+            " WHERE job_id = ? AND table_name = ?", diffKeyspace));
+        jobStatusStatement = session.prepare(String.format(
+            " SELECT " +
+            "   job_id,"   +
+            "   bucket,"   +
+            "   table_name,"   +
+            "   completed "   +
+            " FROM %s.job_status" +
+            " WHERE job_id = ? AND bucket = ?", diffKeyspace));
+        jobMismatchesStatement = session.prepare(String.format(
+            " SELECT " +
+            "   job_id," +
+            "   bucket," +
+            "   table_name," +
+            "   mismatching_token," +
+            "   mismatch_type" +
+            " FROM %s.mismatches" +
+            " WHERE job_id = ? AND bucket = ?", diffKeyspace));
+        jobErrorSummaryStatement = session.prepare(String.format(
+            " SELECT " +
+            "   count(start_token) AS error_count," +
+            "   table_name" +
+            " FROM %s.task_errors" +
+            " WHERE job_id = ? AND bucket = ?",
+            diffKeyspace));
+        jobErrorRangesStatement = session.prepare(String.format(
+            " SELECT " +
+            "   bucket,"   +
+            "   table_name,"   +
+            "   start_token,"   +
+            "   end_token"   +
+            " FROM %s.task_errors" +
+            " WHERE job_id = ? AND bucket = ?",
+            diffKeyspace));
+        jobErrorDetailStatement = session.prepare(String.format(
+            " SELECT " +
+            "   table_name,"   +
+            "   error_token"   +
+            " FROM %s.partition_errors" +
+            " WHERE job_id = ? AND bucket = ? AND table_name = ? AND start_token = ? AND end_token = ?", diffKeyspace));
+        jobsStartDateStatement = session.prepare(String.format(
+            " SELECT " +
+            "   job_id" +
+            " FROM %s.job_start_index" +
+            " WHERE job_start_date = ? AND job_start_hour = ?", diffKeyspace));
+        jobsForSourceStatement = session.prepare(String.format(
+            " SELECT " +
+            "   job_id" +
+            " FROM %s.source_cluster_index" +
+            " WHERE source_cluster_name = ?", diffKeyspace));
+        jobsForTargetStatement = session.prepare(String.format(
+            " SELECT " +
+            "   job_id" +
+            " FROM %s.target_cluster_index" +
+            " WHERE target_cluster_name = ?", diffKeyspace));
+        jobsForKeyspaceStatement = session.prepare(String.format(
+            " SELECT " +
+            "   job_id" +
+            " FROM %s.keyspace_index" +
+            " WHERE keyspace_name = ?", diffKeyspace));
+    }
+
+    public List<UUID> fetchRunningJobs() {
+        List<UUID> jobs = new ArrayList<>();
+        ResultSet rs = session.execute(runningJobsStatement.bind());
+        rs.forEach(row -> jobs.add(row.getUUID("job_id")));
+        return jobs;
+    }
+
+    public Collection<JobSummary> fetchJobSummaries(List<UUID> jobIds) {
+        List<ResultSetFuture> futures = Lists.newArrayListWithCapacity(jobIds.size());
+        jobIds.forEach(id -> futures.add(session.executeAsync(jobSummaryStatement.bind(id))));
+
+        // Oldest first
+        SortedSet<JobSummary> summaries = Sets.newTreeSet(JobSummary.COMPARATOR);
+        processFutures(futures, JobSummary::fromRow, summaries::add);
+        return summaries;
+    }
+
+    public JobSummary fetchJobSummary(UUID jobId) {
+        Row row = session.execute(jobSummaryStatement.bind(jobId)).one();
+        if (row == null)
+            throw new RuntimeException(String.format("Job %s not found", jobId));
+        return JobSummary.fromRow(row);
+    }
+
+    public Collection<JobResult> fetchJobResults(UUID jobId) {
+        JobSummary summary = fetchJobSummary(jobId);
+        List<ResultSetFuture> futures = Lists.newArrayListWithCapacity(summary.tables.size());
+        for (String table : summary.tables)
+            futures.add(session.executeAsync(jobResultStatement.bind(jobId, table)));
+
+        SortedSet<JobResult> results = Sets.newTreeSet();
+        processFutures(futures, JobResult::fromRow, results::add);
+        return results;
+    }
+
+    public JobStatus fetchJobStatus(UUID jobId) {
+        JobSummary summary = fetchJobSummary(jobId);
+        List<ResultSetFuture> futures = Lists.newArrayListWithCapacity(summary.buckets);
+
+        for (int i = 0; i < summary.buckets; i++)
+            futures.add(session.executeAsync(jobStatusStatement.bind(jobId, i)));
+
+        Map<String, Long> completedByTable = Maps.newHashMapWithExpectedSize(summary.tables.size());
+        processFutures(futures, row -> completedByTable.merge(row.getString("table_name"),
+                                                              row.getLong("completed"),
+                                                              Long::sum));
+        return new JobStatus(jobId, completedByTable);
+    }
+
+    public JobMismatches fetchMismatches(UUID jobId) {
+        JobSummary summary = fetchJobSummary(jobId);
+        List<ResultSetFuture> futures = Lists.newArrayListWithCapacity(summary.buckets);
+
+        for (int i = 0; i < summary.buckets; i++)
+            futures.add(session.executeAsync(jobMismatchesStatement.bind(jobId, i)));
+
+        Map<String, List<Mismatch>> mismatchesByTable = Maps.newHashMapWithExpectedSize(summary.tables.size());
+        processFutures(futures, row -> mismatchesByTable.merge(row.getString("table_name"),
+                                                               Lists.newArrayList(new Mismatch(row.getString("mismatching_token"),
+                                                                                               row.getString("mismatch_type"))),
+                                                               (l1, l2) -> { l1.addAll(l2); return l1;}));
+        return new JobMismatches(jobId, mismatchesByTable);
+    }
+
+    public JobErrorSummary fetchErrorSummary(UUID jobId) {
+        JobSummary summary = fetchJobSummary(jobId);
+        List<ResultSetFuture> futures = Lists.newArrayListWithCapacity(summary.buckets);
+
+        for (int i = 0; i < summary.buckets; i++)
+            futures.add(session.executeAsync(jobErrorSummaryStatement.bind(jobId, i)));
+
+        Map<String, Long> errorCountByTable = Maps.newHashMapWithExpectedSize(summary.tables.size());
+        processFutures(futures, row -> {
+            String table = row.getString("table_name");
+            if (null != table) {
+                errorCountByTable.merge(row.getString("table_name"),
+                                        row.getLong("error_count"),
+                                        Long::sum);
+            }
+        });
+        return new JobErrorSummary(jobId, errorCountByTable);
+    }
+
+    public JobErrorRanges fetchErrorRanges(UUID jobId) {
+        JobSummary summary = fetchJobSummary(jobId);
+        List<ResultSetFuture> futures = Lists.newArrayListWithCapacity(summary.buckets);
+
+        for (int i = 0; i < summary.buckets; i++)
+            futures.add(session.executeAsync(jobErrorRangesStatement.bind(jobId, i)));
+
+        Map<String, List<Range>> errorRangesByTable = Maps.newHashMapWithExpectedSize(summary.tables.size());
+        processFutures(futures, row -> errorRangesByTable.merge(row.getString("table_name"),
+                                                                Lists.newArrayList(new Range(row.getString("start_token"),
+                                                                                             row.getString("end_token"))),
+                                                                (l1, l2) -> { l1.addAll(l2); return l1;}));
+        return new JobErrorRanges(jobId, errorRangesByTable);
+    }
+
+    public JobErrorDetail fetchErrorDetail(UUID jobId) {
+        JobSummary summary = fetchJobSummary(jobId);
+        List<ResultSetFuture> rangeFutures = Lists.newArrayListWithCapacity(summary.buckets);
+
+        for (int i = 0; i < summary.buckets; i++ )
+            rangeFutures.add(session.executeAsync(jobErrorRangesStatement.bind(jobId, i)));
+
+        List<ResultSetFuture> errorFutures = Lists.newArrayList();
+        processFutures(rangeFutures,
+                       row -> session.executeAsync(jobErrorDetailStatement.bind(jobId,
+                                                                                row.getInt("bucket"),
+                                                                                row.getString("table_name"),
+                                                                                row.getString("start_token"),
+                                                                                row.getString("end_token"))),
+                       errorFutures::add);
+        Map<String, List<String>> errorsByTable = Maps.newHashMapWithExpectedSize(summary.tables.size());
+        processFutures(errorFutures,
+                       row -> errorsByTable.merge(row.getString("table_name"),
+                                                  Lists.newArrayList(row.getString("error_token")),
+                                                  (l1, l2) -> { l1.addAll(l2); return l1;}));
+        return new JobErrorDetail(jobId, errorsByTable);
+    }
+
+    public Collection<JobSummary> fetchJobsStartedBetween(DateTime start, DateTime end) {
+        int days = Days.daysBetween(start, end).getDays();
+        List<ResultSetFuture> idFutures = Lists.newArrayListWithCapacity(days * 24);
+        for (int i = 0; i <= days; i++) {
+            DateTime date = start.plusDays(i);
+            LocalDate ld = LocalDate.fromYearMonthDay(date.getYear(), date.getMonthOfYear(), date.getDayOfMonth());
+            for (int j = 0; j <= 23; j++) {
+                idFutures.add(session.executeAsync(jobsStartDateStatement.bind(ld, j)));
+            }
+        }
+
+        List<ResultSetFuture> jobFutures = Lists.newArrayList();
+        processFutures(idFutures,
+                       row -> session.executeAsync(jobSummaryStatement.bind(row.getUUID("job_id"))),
+                       jobFutures::add);
+
+        SortedSet<JobSummary> jobs = Sets.newTreeSet(JobSummary.COMPARATOR.reversed());
+        processFutures(jobFutures, JobSummary::fromRow, jobs::add);
+        return jobs;
+    }
+
+    public Collection<JobSummary> fetchJobsForSourceCluster(String sourceClusterName) {
+        ResultSet jobIds = session.execute(jobsForSourceStatement.bind(sourceClusterName));
+        List<ResultSetFuture> futures = Lists.newArrayList();
+        jobIds.forEach(row -> futures.add(session.executeAsync(jobSummaryStatement.bind(row.getUUID("job_id")))));
+
+
+        SortedSet<JobSummary> jobs = Sets.newTreeSet(JobSummary.COMPARATOR.reversed());
+        processFutures(futures, JobSummary::fromRow, jobs::add);
+        return jobs;
+    }
+
+    public Collection<JobSummary> fetchJobsForTargetCluster(String targetClusterName) {
+        ResultSet jobIds = session.execute(jobsForTargetStatement.bind(targetClusterName));
+        List<ResultSetFuture> futures = Lists.newArrayList();
+        jobIds.forEach(row -> futures.add(session.executeAsync(jobSummaryStatement.bind(row.getUUID("job_id")))));
+
+        // most recent first
+        SortedSet<JobSummary> jobs = Sets.newTreeSet(JobSummary.COMPARATOR.reversed());
+        processFutures(futures, JobSummary::fromRow, jobs::add);
+        return jobs;
+    }
+
+    public Collection<JobSummary> fetchJobsForKeyspace(String keyspace) {
+        ResultSet jobIds = session.execute(jobsForKeyspaceStatement.bind(keyspace));
+        List<ResultSetFuture> futures = Lists.newArrayList();
+        jobIds.forEach(row -> futures.add(session.executeAsync(jobSummaryStatement.bind(row.getUUID("job_id")))));
+
+        // most recent first
+        SortedSet<JobSummary> jobs = Sets.newTreeSet(JobSummary.COMPARATOR.reversed());
+        processFutures(futures, JobSummary::fromRow, jobs::add);
+        return jobs;
+    }
+
+    private void processFutures(List<ResultSetFuture> futures, Consumer<Row> consumer) {
+        processFutures(futures, Function.identity(), consumer);
+    }
+
+    private <T> void processFutures(List<ResultSetFuture> futures, Function<Row, T> transform, Consumer<T> consumer) {
+        Consumer<ResultSet> resultConsumer = resultSet -> resultSet.forEach(row -> consumer.accept(transform.apply(row)));
+        futures.forEach(f -> getResultsSafely(f).ifPresent(resultConsumer));
+    }
+
+    private Optional<ResultSet> getResultsSafely(ResultSetFuture future) {
+        try {
+            return Optional.ofNullable(future.get(QUERY_TIMEOUT_MS, TimeUnit.MILLISECONDS));
+        } catch (Exception e) {
+            logger.warn("Error getting result from async query", e);
+            return Optional.empty();
+        }
+    }
+
+    public void close()
+    {
+        session.close();
+        cluster.close();
+    }
+
+    public static class Range {
+        final String start;
+        final String end;
+
+        public Range(String start, String end) {
+            this.start = start;
+            this.end = end;
+        }
+    }
+
+    public static class JobErrorSummary {
+        final UUID jobId;
+        final Map<String, Long> errorsByTable;
+
+        public JobErrorSummary(UUID jobId, Map<String, Long> errorsByTable) {
+            this.jobId = jobId;
+            this.errorsByTable = errorsByTable;
+        }
+    }
+
+    public static class JobErrorRanges {
+        final UUID jobId;
+        final Map<String, List<Range>> rangesByTable;
+
+        public JobErrorRanges(UUID jobId, Map<String, List<Range>> rangesByTable) {
+            this.jobId = jobId;
+            this.rangesByTable = rangesByTable;
+        }
+    }
+
+    public static class JobErrorDetail {
+        final UUID jobId;
+        final Map<String, List<String>> errorsByTable;
+
+        public JobErrorDetail(UUID jobId, Map<String, List<String>> errorsByTable) {
+            this.jobId = jobId;
+            this.errorsByTable = errorsByTable;
+        }
+    }
+
+    public static class JobMismatches {
+
+        final UUID jobId;
+        final Map<String, List<Mismatch>> mismatchesByTable;
+
+        public JobMismatches(UUID jobId, Map<String, List<Mismatch>> mismatchesByTable) {
+            this.jobId = jobId;
+            this.mismatchesByTable = mismatchesByTable;
+        }
+    }
+
+    public static class Mismatch {
+        final String token;
+        final String type;
+
+        public Mismatch(String token, String type) {
+            this.token = token;
+            this.type = type;
+        }
+    }
+
+    public static class JobStatus {
+
+        final UUID jobId;
+        final Map<String, Long> completedByTable;
+
+        public JobStatus(UUID jobId, Map<String, Long> completedByTable) {
+            this.jobId = jobId;
+            this.completedByTable = completedByTable;
+        }
+    }
+
+    public static class JobResult implements Comparable<JobResult> {
+
+        final UUID jobId;
+        final String table;
+        final long matchedPartitions;
+        final long mismatchedPartitions;
+        final long matchedRows;
+        final long matchedValues;
+        final long mismatchedValues;
+        final long onlyInSource;
+        final long onlyInTarget;
+        final long skippedPartitions;
+
+        public JobResult(UUID jobId,
+                         String table,
+                         long matchedPartitions,
+                         long mismatchedPartitions,
+                         long matchedRows,
+                         long matchedValues,
+                         long mismatchedValues,
+                         long onlyInSource,
+                         long onlyInTarget,
+                         long skippedPartitions) {
+            this.jobId = jobId;
+            this.table = table;
+            this.matchedPartitions = matchedPartitions;
+            this.mismatchedPartitions = mismatchedPartitions;
+            this.matchedRows = matchedRows;
+            this.matchedValues = matchedValues;
+            this.mismatchedValues = mismatchedValues;
+            this.onlyInSource = onlyInSource;
+            this.onlyInTarget = onlyInTarget;
+            this.skippedPartitions = skippedPartitions;
+        }
+
+        static JobResult fromRow(Row row) {
+            return new JobResult(row.getUUID("job_id"),
+                                 row.getString("table_name"),
+                                 row.getLong("matched_partitions"),
+                                 row.getLong("mismatched_partitions"),
+                                 row.getLong("matched_rows"),
+                                 row.getLong("matched_values"),
+                                 row.getLong("mismatched_values"),
+                                 row.getLong("partitions_only_in_source"),
+                                 row.getLong("partitions_only_in_target"),
+                                 row.getLong("skipped_partitions"));
+        }
+
+        public int compareTo(@NotNull JobResult other) {
+            return this.table.compareTo(other.table);
+        }
+    }
+
+    public static class JobSummary {
+
+        public static final Comparator<JobSummary> COMPARATOR = Comparator.comparing(j -> j.startTime);
+
+        final UUID jobId;
+        final int buckets;
+        final String keyspace;
+        final List<String> tables;
+        final String sourceClusterName;
+        final String sourceClusterDesc;
+        final String targetClusterName;
+        final String targetClusterDesc;
+        final int tasks;
+        final String start;
+
+        // private so it isn't included in json serialization
+        private final DateTime startTime;
+
+        private JobSummary(UUID jobId,
+                           DateTime startTime,
+                           int buckets,
+                           String keyspace,
+                           List<String> tables,
+                           String sourceClusterName,
+                           String sourceClusterDesc,
+                           String targetClusterName,
+                           String targetClusterDesc,
+                           int tasks)
+        {
+            this.jobId = jobId;
+            this.startTime = startTime;
+            this.start = startTime.toString();
+            this.buckets = buckets;
+            this.keyspace = keyspace;
+            this.tables = tables;
+            this.sourceClusterName = sourceClusterName;
+            this.sourceClusterDesc = sourceClusterDesc;
+            this.targetClusterName = targetClusterName;
+            this.targetClusterDesc = targetClusterDesc;
+            this.tasks = tasks;
+        }
+
+        static JobSummary fromRow(Row row) {
+            return new JobSummary(row.getUUID("job_id"),
+                                  new DateTime(UUIDs.unixTimestamp(row.getUUID("job_start_time")), DateTimeZone.UTC),
+                                  row.getInt("buckets"),
+                                  row.getString("keyspace_name"),
+                                  row.getList("table_names", String.class),
+                                  row.getString("source_cluster_name"),
+                                  row.getString("source_cluster_desc"),
+                                  row.getString("target_cluster_name"),
+                                  row.getString("target_cluster_desc"),
+                                  row.getInt("total_tasks"));
+        }
+    }
+}
diff --git a/api-server/src/main/resources/log4j2.xml b/api-server/src/main/resources/log4j2.xml
new file mode 100644
index 0000000..9e401d6
--- /dev/null
+++ b/api-server/src/main/resources/log4j2.xml
@@ -0,0 +1,32 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ 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.
+  -->
+
+<Configuration>
+  <Appenders>
+    <Console name="Console" target="SYSTEM_OUT">
+      <PatternLayout pattern="%d [%t] %-5level %logger{36} - %msg%n"/>
+    </Console>
+  </Appenders>
+  <Loggers>
+    <Root level="info">
+      <AppenderRef ref="Console"/>
+    </Root>
+  </Loggers>
+</Configuration>
diff --git a/common/pom.xml b/common/pom.xml
new file mode 100644
index 0000000..f73c4f5
--- /dev/null
+++ b/common/pom.xml
@@ -0,0 +1,54 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ 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.
+  -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
+         http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+      <groupId>org.apache.cassandra.diff</groupId>
+      <artifactId>diff</artifactId>
+      <version>0.1-SNAPSHOT</version>
+    </parent>
+
+    <modelVersion>4.0.0</modelVersion>
+    <artifactId>common</artifactId>
+    <packaging>jar</packaging>
+
+    <dependencies>
+
+        <dependency>
+            <groupId>org.yaml</groupId>
+            <artifactId>snakeyaml</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>com.datastax.cassandra</groupId>
+            <artifactId>cassandra-driver-core</artifactId>
+        </dependency>
+
+    </dependencies>
+
+</project>
\ No newline at end of file
diff --git a/common/src/main/java/org/apache/cassandra/diff/ClusterProvider.java b/common/src/main/java/org/apache/cassandra/diff/ClusterProvider.java
new file mode 100644
index 0000000..6c4e3e4
--- /dev/null
+++ b/common/src/main/java/org/apache/cassandra/diff/ClusterProvider.java
@@ -0,0 +1,46 @@
+/*
+ * 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.diff;
+
+import java.io.Serializable;
+import java.util.Map;
+
+import com.datastax.driver.core.Cluster;
+
+public interface ClusterProvider extends Serializable {
+    String CLUSTER_PROVIDER_CLASS = "impl";
+
+    void initialize(Map<String, String> conf, String identifier);
+    Cluster getCluster();
+    String getClusterName();
+
+    static ClusterProvider getProvider(Map<String, String> conf, String identifier) {
+        String providerImpl = conf.get(CLUSTER_PROVIDER_CLASS);
+        ClusterProvider provider;
+        try {
+            provider = (ClusterProvider)(Class.forName(providerImpl)).newInstance();
+            provider.initialize(conf, identifier);
+        } catch (Exception e) {
+            throw new RuntimeException("Could not instantiate ClusterProvider class: " + providerImpl +" " + conf, e);
+        }
+        return provider;
+    }
+
+}
diff --git a/common/src/main/java/org/apache/cassandra/diff/ContactPointsClusterProvider.java b/common/src/main/java/org/apache/cassandra/diff/ContactPointsClusterProvider.java
new file mode 100644
index 0000000..b633feb
--- /dev/null
+++ b/common/src/main/java/org/apache/cassandra/diff/ContactPointsClusterProvider.java
@@ -0,0 +1,80 @@
+/*
+ * 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.diff;
+
+import java.util.Map;
+
+import com.datastax.driver.core.Cluster;
+
+public class ContactPointsClusterProvider implements ClusterProvider {
+    private static final String PREFIX = "diff.cluster";
+    private static final String USERNAME_KEY = "cql_user";
+    private static final String PASSWORD_KEY = "cql_password";
+    private static final String CONTACT_POINTS_KEY = "contact_points";
+    private static final String PORT_KEY = "port";
+    private static final String CLUSTER_KEY = "name";
+    private static final String DC_KEY = "dc";
+
+    private String user;
+    private String password;
+    private String name;
+    private String dc;
+    private String[] contactPoints;
+    private int port;
+
+    public Cluster getCluster() {
+        return newCluster();
+    }
+
+    public void initialize(Map<String, String> conf, String identifier) {
+        user = getEnv(identifier, USERNAME_KEY);
+        password = getEnv(identifier, PASSWORD_KEY);
+        name = conf.get(CLUSTER_KEY);
+        dc = conf.get(DC_KEY);
+        contactPoints = conf.get(CONTACT_POINTS_KEY).split(",");
+        port = Integer.parseInt(conf.getOrDefault(PORT_KEY, "9042"));
+    }
+
+    private synchronized Cluster newCluster() {
+        // TODO add policies etc
+        Cluster.Builder builder = Cluster.builder()
+                                         .addContactPoints(contactPoints)
+                                         .withPort(port);
+
+        if (user != null)
+            builder.withCredentials(user, password);
+
+        return builder.build();
+    }
+
+    public String getClusterName() {
+        return name;
+    }
+
+    public String toString() {
+        return String.format("ContactPoints Cluster: name=%s, dc=%s, contact points= [%s]",
+                             "name", dc, String.join(",'", contactPoints));
+    }
+
+    static String getEnv(String identifier, String propName) {
+        return System.getenv(String.format("%s.%s.%s", PREFIX, identifier, propName));
+    }
+
+}
diff --git a/common/src/main/java/org/apache/cassandra/diff/JobConfiguration.java b/common/src/main/java/org/apache/cassandra/diff/JobConfiguration.java
new file mode 100644
index 0000000..f0cbb36
--- /dev/null
+++ b/common/src/main/java/org/apache/cassandra/diff/JobConfiguration.java
@@ -0,0 +1,60 @@
+/*
+ * 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.diff;
+
+import java.io.Serializable;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.UUID;
+
+public interface JobConfiguration extends Serializable {
+    String keyspace();
+
+    List<String> tables();
+
+    int splits();
+
+    int buckets();
+
+    // rate limit is provided as a global limit - this is how many q/s we guess that the src clusters can take in total
+    int rateLimit();
+
+    default SpecificTokens specificTokens() {
+        return SpecificTokens.NONE;
+    }
+
+    Optional<UUID> jobId();
+
+    int tokenScanFetchSize();
+
+    int partitionReadFetchSize();
+
+    int readTimeoutMillis();
+
+    double reverseReadProbability();
+
+    String consistencyLevel();
+
+    MetadataKeyspaceOptions metadataOptions();
+
+    Map<String, String> clusterConfig(String identifier);
+
+}
diff --git a/common/src/main/java/org/apache/cassandra/diff/MetadataKeyspaceOptions.java b/common/src/main/java/org/apache/cassandra/diff/MetadataKeyspaceOptions.java
new file mode 100644
index 0000000..b3e9eb0
--- /dev/null
+++ b/common/src/main/java/org/apache/cassandra/diff/MetadataKeyspaceOptions.java
@@ -0,0 +1,29 @@
+/*
+ * 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.diff;
+
+import java.io.Serializable;
+
+public class MetadataKeyspaceOptions implements Serializable {
+    public String keyspace = "cassandradiff";
+    public String replication = "{'class':'SimpleStrategy', 'replication_factor':'1'}";
+    public int ttl = 60 * 60 * 24 * 365;
+    public boolean should_init = false;
+}
diff --git a/common/src/main/java/org/apache/cassandra/diff/SpecificTokens.java b/common/src/main/java/org/apache/cassandra/diff/SpecificTokens.java
new file mode 100644
index 0000000..5400a54
--- /dev/null
+++ b/common/src/main/java/org/apache/cassandra/diff/SpecificTokens.java
@@ -0,0 +1,61 @@
+/*
+ * 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.diff;
+
+import java.io.Serializable;
+import java.math.BigInteger;
+import java.util.Collections;
+import java.util.Set;
+import java.util.function.Predicate;
+
+import com.google.common.collect.ImmutableSet;
+
+public class SpecificTokens implements Serializable, Predicate<BigInteger> {
+
+    public static final SpecificTokens NONE = new SpecificTokens(Collections.emptySet(), Modifier.REJECT);
+
+    public enum Modifier {REJECT, ACCEPT}
+
+    public final ImmutableSet<BigInteger> tokens;
+    public final Modifier modifier;
+
+    public SpecificTokens(Set<BigInteger> tokens, Modifier modifier) {
+        this.tokens = ImmutableSet.copyOf(tokens);
+        this.modifier = modifier;
+    }
+
+    public boolean test(BigInteger token) {
+        if (tokens.isEmpty())
+            return true;
+
+        // if this represents a list of tokens to accept, then this token is allowed
+        // only if it is present in the list. If this is a list of tokens to reject,
+        // then the opposite is true, only allow the token if it is not in the list.
+        return (modifier == Modifier.ACCEPT) == tokens.contains(token);
+    }
+
+    public boolean isEmpty() {
+        return tokens.isEmpty();
+    }
+
+    public String toString() {
+        return String.format("SpecificTokens: [ %s, %s ]", modifier, tokens);
+    }
+}
diff --git a/common/src/main/java/org/apache/cassandra/diff/YamlJobConfiguration.java b/common/src/main/java/org/apache/cassandra/diff/YamlJobConfiguration.java
new file mode 100644
index 0000000..69fc28c
--- /dev/null
+++ b/common/src/main/java/org/apache/cassandra/diff/YamlJobConfiguration.java
@@ -0,0 +1,154 @@
+/*
+ * 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.diff;
+
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
+import java.math.BigInteger;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.UUID;
+
+import org.yaml.snakeyaml.Yaml;
+import org.yaml.snakeyaml.constructor.CustomClassLoaderConstructor;
+
+public class YamlJobConfiguration implements JobConfiguration {
+    public int splits = 10000;
+    public String keyspace;
+    public List<String> tables;
+    public int buckets = 100;
+    public int rate_limit = 10000;
+    public String job_id = null;
+    public int token_scan_fetch_size;
+    public int partition_read_fetch_size;
+    public int read_timeout_millis;
+    public double reverse_read_probability;
+    public String consistency_level = "ALL";
+    public MetadataKeyspaceOptions metadata_options;
+    public Map<String, Map<String, String>> cluster_config;
+    public String specific_tokens = null;
+    public String disallowed_tokens = null;
+
+    public static YamlJobConfiguration load(String file) {
+        Yaml yaml = new Yaml(new CustomClassLoaderConstructor(YamlJobConfiguration.class,
+                                                              Thread.currentThread().getContextClassLoader()));
+        try {
+            return yaml.loadAs(new FileInputStream(file), YamlJobConfiguration.class);
+        } catch (FileNotFoundException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    public String keyspace() {
+        return keyspace;
+    }
+
+    public List<String> tables() {
+        return tables;
+    }
+
+    public int splits() {
+        return splits;
+    }
+
+    public int buckets() {
+        return buckets;
+    }
+
+    public int rateLimit() {
+        return rate_limit;
+    }
+
+    public Optional<UUID> jobId() {
+        return job_id == null ? Optional.empty() : Optional.of(UUID.fromString(job_id));
+    }
+
+    public int tokenScanFetchSize() {
+        return token_scan_fetch_size;
+    }
+
+    public int partitionReadFetchSize() {
+        return partition_read_fetch_size;
+    }
+
+    public int readTimeoutMillis() {
+        return read_timeout_millis;
+    }
+
+    public double reverseReadProbability() {
+        return reverse_read_probability;
+    }
+
+    public String consistencyLevel() {
+        return consistency_level;
+    }
+
+    public MetadataKeyspaceOptions metadataOptions() {
+        return metadata_options;
+    }
+
+    public Map<String, String> clusterConfig(String identifier) {
+        return cluster_config.get(identifier);
+    }
+
+    public SpecificTokens specificTokens() {
+
+        if (disallowed_tokens != null && specific_tokens != null)
+            throw new RuntimeException("Cannot specify both disallowed and specific tokens");
+
+        if (disallowed_tokens != null) {
+            return new SpecificTokens(toTokens(disallowed_tokens), SpecificTokens.Modifier.REJECT);
+        } else if (specific_tokens != null) {
+            return new SpecificTokens(toTokens(specific_tokens), SpecificTokens.Modifier.ACCEPT);
+        }
+        return SpecificTokens.NONE;
+    }
+
+    public String toString() {
+        return "YamlJobConfiguration{" +
+               "splits=" + splits +
+               ", keyspace='" + keyspace + '\'' +
+               ", tables=" + tables +
+               ", buckets=" + buckets +
+               ", rate_limit=" + rate_limit +
+               ", job_id='" + job_id + '\'' +
+               ", token_scan_fetch_size=" + token_scan_fetch_size +
+               ", partition_read_fetch_size=" + partition_read_fetch_size +
+               ", read_timeout_millis=" + read_timeout_millis +
+               ", reverse_read_probability=" + reverse_read_probability +
+               ", consistency_level='" + consistency_level + '\'' +
+               ", metadata_options=" + metadata_options +
+               ", cluster_config=" + cluster_config +
+               '}';
+    }
+
+    private static Set<BigInteger> toTokens(String str) {
+        Set<BigInteger> tokens = new HashSet<>();
+        for (String token : str.split(",")) {
+            token = token.trim();
+            tokens.add(new BigInteger(token));
+        }
+        return tokens;
+    }
+
+}
diff --git a/pom.xml b/pom.xml
new file mode 100644
index 0000000..fce7cc1
--- /dev/null
+++ b/pom.xml
@@ -0,0 +1,108 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ 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.
+  -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 
+		 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <groupId>org.apache.cassandra.diff</groupId>
+    <artifactId>diff</artifactId>
+    <packaging>pom</packaging>
+    <version>0.1-SNAPSHOT</version>
+    <modules>
+      <module>common</module>
+      <module>spark-job</module>
+      <module>api-server</module>
+    </modules>
+
+    <properties>
+      <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+    </properties>
+
+    <dependencyManagement>
+      <dependencies>
+
+        <dependency>
+          <groupId>com.datastax.cassandra</groupId>
+          <artifactId>cassandra-driver-core</artifactId>
+          <version>3.7.1</version>
+          <scope>compile</scope>
+        </dependency>
+
+        <dependency>
+          <groupId>org.yaml</groupId>
+          <artifactId>snakeyaml</artifactId>
+          <version>1.24</version>
+          <scope>compile</scope>
+        </dependency>
+
+        <dependency>
+          <groupId>com.google.guava</groupId>
+          <artifactId>guava</artifactId>
+          <version>28.0-jre</version>
+          <scope>compile</scope>
+        </dependency>
+
+        <dependency>
+          <groupId>org.slf4j</groupId>
+          <artifactId>slf4j-api</artifactId>
+          <version>1.7.28</version>
+          <scope>compile</scope>
+        </dependency>
+
+        <dependency>
+          <groupId>org.jetbrains</groupId>
+          <artifactId>annotations</artifactId>
+          <version>17.0.0</version>
+          <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+          <groupId>org.apache.logging.log4j</groupId>
+          <artifactId>log4j-slf4j-impl</artifactId>
+          <version>2.12.1</version>
+          <scope>runtime</scope>
+        </dependency>
+
+        <dependency>
+          <groupId>junit</groupId>
+          <artifactId>junit</artifactId>
+          <version>4.12</version>
+          <scope>test</scope>
+        </dependency>
+
+      </dependencies>
+    </dependencyManagement>
+    <build>
+      <plugins>
+        <plugin>
+          <groupId>org.apache.maven.plugins</groupId>
+          <artifactId>maven-compiler-plugin</artifactId>
+          <version>3.6.1</version>
+          <configuration>
+            <source>1.8</source>
+            <target>1.8</target>
+          </configuration>
+        </plugin>
+      </plugins>
+    </build>
+</project>
diff --git a/spark-job/localconfig.yaml b/spark-job/localconfig.yaml
new file mode 100644
index 0000000..d4f5630
--- /dev/null
+++ b/spark-job/localconfig.yaml
@@ -0,0 +1,51 @@
+# Keyspace to diff
+keyspace: keyspace1
+# List of tables to diff
+tables:
+  - standard1
+
+# This is how many parts we split the full token range in.
+# Each of these splits is then compared between the clusters
+splits: 10000
+
+# Number of buckets - splits / buckets should be under 100k to avoid wide partitions when storing the metadata
+buckets: 100
+
+# global rate limit - this is how many q/s you think the target clusters can handle
+rate_limit: 10000
+
+# optional job id - if restarting a job, set the correct job_id here to avoid re-diffing old splits
+# job_id: 4e2c6c6b-bed7-4c4e-bd4c-28bef89c3cef
+
+# Fetch size to use for the query fetching the tokens in the cluster
+token_scan_fetch_size: 1000
+# Fetch size to use for the queries fetching the rows of each partition
+partition_read_fetch_size: 1000
+
+read_timeout_millis: 10000
+reverse_read_probability: 0.5
+consistency_level: ALL
+metadata_options:
+  keyspace: cassandradiff
+  replication: "{'class':'SimpleStrategy', 'replication_factor':'1'}"
+  ttl: 31536000
+  should_init: true
+cluster_config:
+  source:
+    impl: "org.apache.cassandra.diff.ContactPointsClusterProvider"
+    name: "local_test_1"
+    contact_points: "127.0.0.1"
+    port: "9042"
+    dc: "datacenter1"
+  target:
+    impl: "org.apache.cassandra.diff.ContactPointsClusterProvider"
+    name: "local_test_2"
+    contact_points: "127.0.0.1"
+    port: "9043"
+    dc: "datacenter1"
+  metadata:
+    impl: "org.apache.cassandra.diff.ContactPointsClusterProvider"
+    name: "local_test"
+    contact_points: "127.0.0.1"
+    port: "9042"
+    dc: "datacenter1"
diff --git a/spark-job/pom.xml b/spark-job/pom.xml
new file mode 100644
index 0000000..4d4c8fc
--- /dev/null
+++ b/spark-job/pom.xml
@@ -0,0 +1,101 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ 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.
+  -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
+		 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <groupId>org.apache.cassandra.diff</groupId>
+        <artifactId>diff</artifactId>
+        <version>0.1-SNAPSHOT</version>
+    </parent>
+
+    <modelVersion>4.0.0</modelVersion>
+    <artifactId>spark-job</artifactId>
+    <packaging>jar</packaging>
+
+    <dependencies>
+
+        <!-- compile dependencies -->
+        <dependency>
+            <groupId>org.apache.cassandra.diff</groupId>
+            <artifactId>common</artifactId>
+            <version>${project.parent.version}</version>
+        </dependency>
+
+        <!-- provided dependencies -->
+        <dependency>
+          <groupId>org.jetbrains</groupId>
+          <artifactId>annotations</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.spark</groupId>
+            <artifactId>spark-core_2.11</artifactId>
+            <version>2.3.3</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.spark</groupId>
+            <artifactId>spark-sql_2.11</artifactId>
+            <version>2.3.2</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <!-- test dependencies -->
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+        </dependency>
+
+    </dependencies>
+
+    <build>
+        <plugins>
+
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-shade-plugin</artifactId>
+                <version>3.2.1</version>
+                <executions>
+                    <!-- Run shade goal on package phase -->
+                    <execution>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>shade</goal>
+                        </goals>
+                    </execution>
+                </executions>
+                <configuration>
+                    <relocations>
+                        <relocation>
+                            <pattern>com.google</pattern>
+                            <shadedPattern>relocated.com.google</shadedPattern>
+                        </relocation>
+                    </relocations>
+                </configuration>
+            </plugin>
+
+        </plugins>
+    </build>
+
+</project>
diff --git a/spark-job/src/main/java/org/apache/cassandra/diff/ComparisonExecutor.java b/spark-job/src/main/java/org/apache/cassandra/diff/ComparisonExecutor.java
new file mode 100644
index 0000000..bdd6488
--- /dev/null
+++ b/spark-job/src/main/java/org/apache/cassandra/diff/ComparisonExecutor.java
@@ -0,0 +1,116 @@
+/*
+ * 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.diff;
+
+import java.util.concurrent.*;
+import java.util.function.Consumer;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.util.concurrent.*;
+
+import com.codahale.metrics.Gauge;
+import com.codahale.metrics.MetricRegistry;
+
+/**
+ * Wrapper for an ExecutorService which provides backpressure by blocking on submission when its
+ * task queue is full.
+ *
+ * The internal ListeningExecutorService is instantiated with an unbounded work queue, but
+ * this class uses a Semaphore to ensure that this queue cannot grow unreasonably. By default,
+ * the Semaphore has 2x as many permits as the executor thread pool has threads, so at most
+ * 2 x maxConcurrentTasks may be submitted before producers are blocked. The submit method
+ * also adds success/failure callbacks which return the permits, enabling producers to make
+ * progress at a manageable rate.
+ *
+ * Callers of the submit method also provide a Phaser, which they can use to ensure that any
+ * tasks *they themselves have submitted* are completed before they proceed. This allows multiple
+ * callers to submit tasks to the same ComparisonExecutor, but only wait for their own to complete
+ * before moving onto the next stage of processing. Managing the increment and decrement of pending
+ * tasks via the Phaser is handled transparently by ComparisonExecutor, so callers should not do
+ * this externally.
+ *
+ * Submitters also provide callbacks to be run on either successful execution or failure of the
+ * task. These callbacks are executed on the same thread as the task itself, which callers should bear
+ * in mind when constructing them.
+ *
+ */
+public class ComparisonExecutor {
+
+    private final ListeningExecutorService executor;
+    private final Semaphore semaphore;
+
+    static ComparisonExecutor newExecutor(int maxConcurrentTasks, MetricRegistry metricRegistry) {
+        return new ComparisonExecutor(
+            MoreExecutors.listeningDecorator(
+                Executors.newFixedThreadPool(maxConcurrentTasks,
+                                             new ThreadFactoryBuilder().setNameFormat("partition-comparison-%d")
+                                                                       .setDaemon(true)
+                                                                       .build())),
+            maxConcurrentTasks * 2,
+            metricRegistry);
+    }
+
+    @VisibleForTesting
+    ComparisonExecutor(ListeningExecutorService executor, int maxTasks, MetricRegistry metrics) {
+        this.executor = executor;
+        this.semaphore = new Semaphore(maxTasks);
+        if (metrics != null) {
+            metrics.register("BlockedTasks", (Gauge) semaphore::getQueueLength);
+            metrics.register("AvailableSlots", (Gauge) semaphore::availablePermits);
+        }
+    }
+
+    public <T> void submit(final Callable<T> callable,
+                           final Consumer<T> onSuccess,
+                           final Consumer<Throwable> onError,
+                           final Phaser phaser) {
+
+        phaser.register();
+        semaphore.acquireUninterruptibly();
+        try {
+            Futures.addCallback(executor.submit(callable), new FutureCallback<T>() {
+                public void onSuccess(T result) {
+                    fireThenReleaseAndArrive(onSuccess, result, phaser);
+                }
+
+                public void onFailure(Throwable t) {
+                    fireThenReleaseAndArrive(onError, t, phaser);
+                }
+            }, MoreExecutors.directExecutor());
+
+        } catch (RejectedExecutionException e) {
+            fireThenReleaseAndArrive(onError, e, phaser);
+        }
+
+    }
+
+    private <T> void fireThenReleaseAndArrive(Consumer<T> callback, T argument, Phaser phaser) {
+        try {
+            callback.accept(argument);
+        } finally {
+            semaphore.release();
+            phaser.arriveAndDeregister();
+        }
+    }
+
+    public void shutdown() {
+        executor.shutdown();
+    }
+}
diff --git a/spark-job/src/main/java/org/apache/cassandra/diff/DiffCluster.java b/spark-job/src/main/java/org/apache/cassandra/diff/DiffCluster.java
new file mode 100644
index 0000000..4e108e3
--- /dev/null
+++ b/spark-job/src/main/java/org/apache/cassandra/diff/DiffCluster.java
@@ -0,0 +1,245 @@
+/*
+ * 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.diff;
+
+import java.math.BigInteger;
+import java.util.*;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import com.google.common.collect.AbstractIterator;
+import com.google.common.util.concurrent.*;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.datastax.driver.core.*;
+import com.datastax.driver.core.querybuilder.*;
+import org.jetbrains.annotations.NotNull;
+
+import static org.apache.cassandra.diff.DiffContext.cqlizedString;
+import static com.datastax.driver.core.querybuilder.QueryBuilder.*;
+
+public class DiffCluster implements AutoCloseable
+{
+    private final static Logger logger = LoggerFactory.getLogger(DiffCluster.class);
+
+    public enum Type {SOURCE, TARGET}
+
+    private final Map<String, PreparedStatement[]> preparedStatements = new HashMap<>();
+    private final ConsistencyLevel consistencyLevel;
+    public final Cluster cluster;
+    private final Session session;
+    private final TokenHelper tokenHelper;
+    public final String keyspace;
+    public final List<BigInteger> tokenList;
+
+    public final RateLimiter getPartitionRateLimiter;
+    public final Type clusterId;
+    private final int tokenScanFetchSize;
+    private final int partitionReadFetchSize;
+    private final int readTimeoutMillis;
+
+    private final AtomicBoolean stopped = new AtomicBoolean(false);
+
+    public DiffCluster(Type clusterId,
+                       Cluster cluster,
+                       String keyspace,
+                       ConsistencyLevel consistencyLevel,
+                       RateLimiter getPartitionRateLimiter,
+                       int tokenScanFetchSize,
+                       int partitionReadFetchSize,
+                       int readTimeoutMillis)
+
+    {
+        this.keyspace = keyspace;
+        this.consistencyLevel = consistencyLevel;
+        this.cluster = cluster;
+        this.tokenHelper = TokenHelper.forPartitioner(cluster.getMetadata().getPartitioner());
+        this.clusterId = clusterId;
+        this.tokenList = Collections.emptyList();
+        this.getPartitionRateLimiter = getPartitionRateLimiter;
+        this.session = cluster.connect();
+        this.tokenScanFetchSize = tokenScanFetchSize;
+        this.partitionReadFetchSize = partitionReadFetchSize;
+        this.readTimeoutMillis = readTimeoutMillis;
+    }
+
+    public Iterator<PartitionKey> getPartitionKeys(String table, final BigInteger prevToken, final BigInteger token) {
+        try {
+            return Uninterruptibles.getUninterruptibly(fetchPartitionKeys(table, prevToken, token));
+        }
+        catch (ExecutionException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    private ListenableFuture<Iterator<PartitionKey>> fetchPartitionKeys(String table, final BigInteger prevToken, final BigInteger token) {
+        BoundStatement statement = keyReader(table).bind(tokenHelper.forBindParam(prevToken),
+                                                         tokenHelper.forBindParam(token));
+        statement.setFetchSize(tokenScanFetchSize);
+        statement.setReadTimeoutMillis(readTimeoutMillis);
+        return Futures.transform(session.executeAsync(statement),
+                                 this::toPartitionKeys,
+                                 MoreExecutors.directExecutor());
+    }
+
+    private AbstractIterator<PartitionKey> toPartitionKeys(ResultSet resultSet) {
+        return new AbstractIterator<PartitionKey>() {
+            Iterator<Row> rows = resultSet.iterator();
+
+            protected PartitionKey computeNext() {
+                if (session.isClosed())
+                    throw new RuntimeException("Session was closed, cannot get next partition key");
+
+                if (stopped.get())
+                    throw new RuntimeException("Job was stopped, cannot get next partition key");
+
+                return rows.hasNext() ? new PartitionKey(rows.next()) : endOfData();
+            }
+        };
+    }
+
+    public Iterator<Row> getPartition(TableSpec table, PartitionKey key, boolean shouldReverse) {
+        return readPartition(table.getTable(), key, shouldReverse)
+               .getUninterruptibly()
+               .iterator();
+    }
+
+    private ResultSetFuture readPartition(String table, final PartitionKey key, boolean shouldReverse) {
+        BoundStatement statement = shouldReverse
+                                   ? reverseReader(table).bind(key.getComponents().toArray())
+                                   : forwardReader(table).bind(key.getComponents().toArray());
+        statement.setFetchSize(partitionReadFetchSize);
+        statement.setReadTimeoutMillis(readTimeoutMillis);
+        getPartitionRateLimiter.acquire();
+        return session.executeAsync(statement);
+    }
+
+    public void stop() {
+        stopped.set(true);
+    }
+
+    public void close()
+    {
+        logger.info("Closing cluster {}", this.clusterId);
+        session.closeAsync();
+        cluster.closeAsync();
+    }
+
+    private PreparedStatement keyReader(String table) {
+        return getStatementForTable(table, 0);
+    }
+
+    private PreparedStatement forwardReader(String table) {
+        return getStatementForTable(table, 1);
+    }
+
+    private PreparedStatement reverseReader(String table) {
+        return getStatementForTable(table, 2);
+    }
+
+    private PreparedStatement getStatementForTable(String table, int index) {
+        if (!preparedStatements.containsKey(table)) {
+            synchronized (this) {
+                if (!preparedStatements.containsKey(table)) {
+                    PreparedStatement keyStatement = getKeyStatement(table);
+                    PreparedStatement[] partitionReadStmts = getFullStatement(table);
+                    preparedStatements.put(table, new PreparedStatement[]{ keyStatement ,
+                                                                           partitionReadStmts[0],
+                                                                           partitionReadStmts[1] });
+                }
+            }
+        }
+        return preparedStatements.get(table)[index];
+    }
+
+    private PreparedStatement getKeyStatement(@NotNull String table) {
+        final TableMetadata tableMetadata = session.getCluster()
+                                                   .getMetadata()
+                                                   .getKeyspace(cqlizedString(keyspace))
+                                                   .getTable(cqlizedString(table));
+        String[] partitionKeyColumns = columnNames(tableMetadata.getPartitionKey());
+
+        Select.Selection selection = QueryBuilder.select().distinct().column(token(partitionKeyColumns));
+        for (String column : partitionKeyColumns)
+            selection = selection.column(column);
+
+        BuiltStatement select = selection.from(tableMetadata)
+                                         .where(gt(token(partitionKeyColumns), bindMarker()))
+                                         .and(lte(token(partitionKeyColumns), bindMarker()));
+
+        logger.debug("Partition key/token read CQL : {}", select.toString());
+        return session.prepare(select).setConsistencyLevel(consistencyLevel);
+    }
+
+    private PreparedStatement[] getFullStatement(@NotNull String table) {
+        final TableMetadata tableMetadata = session.getCluster()
+                                                   .getMetadata()
+                                                   .getKeyspace(cqlizedString(keyspace))
+                                                   .getTable(cqlizedString(table));
+        String[] partitionKeyColumns = columnNames(tableMetadata.getPartitionKey());
+        String[] allColumns = columnNames(tableMetadata.getColumns());
+
+        Select.Selection selection = QueryBuilder.select().column(token(partitionKeyColumns));
+        for (String column : allColumns)
+            selection = selection.column(column);
+
+        Select select = selection.from(tableMetadata);
+
+        for (String column : partitionKeyColumns)
+            select.where().and(eq(column, bindMarker()));
+
+        logger.info("Partition forward read CQL : {}", select.toString());
+        PreparedStatement forwardRead = session.prepare(select).setConsistencyLevel(consistencyLevel);
+
+        List<ColumnMetadata> clusteringColumns = tableMetadata.getClusteringColumns();
+        // if the table has no clustering columns a reverse read doesn't make sense
+        // and will never be executed, so just skip preparing the reverse query
+        if (clusteringColumns.isEmpty())
+            return new PreparedStatement[] {forwardRead, null};
+
+        // Depending on DiffContext.reverseReadProbability, we may attempt to read the
+        // partition in reverse order, so prepare a statement for that
+        List<ClusteringOrder> clusteringOrders = tableMetadata.getClusteringOrder();
+        Ordering[] reverseOrdering = new Ordering[clusteringColumns.size()];
+        for (int i=0; i<clusteringColumns.size(); i++) {
+            reverseOrdering[i] = clusteringOrders.get(i) == ClusteringOrder.ASC
+                                 ? desc(clusteringColumns.get(i).getName())
+                                 : asc(clusteringColumns.get(i).getName());
+        }
+
+        select.orderBy(reverseOrdering);
+        logger.info("Partition reverse read CQL : {}", select.toString());
+
+        PreparedStatement reverseRead = session.prepare(select).setConsistencyLevel(consistencyLevel);
+
+        return new PreparedStatement[] {forwardRead, reverseRead};
+    }
+
+    private static String[] columnNames(List<ColumnMetadata> columns) {
+        return columns.stream().map(ColumnMetadata::getName).map(DiffCluster::columnToString).toArray(String[]::new);
+    }
+
+    private static String columnToString(String name)
+    {
+        return '"'+name+'"';
+    }
+}
diff --git a/spark-job/src/main/java/org/apache/cassandra/diff/DiffContext.java b/spark-job/src/main/java/org/apache/cassandra/diff/DiffContext.java
new file mode 100644
index 0000000..5a1a354
--- /dev/null
+++ b/spark-job/src/main/java/org/apache/cassandra/diff/DiffContext.java
@@ -0,0 +1,75 @@
+/*
+ * 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.diff;
+
+import java.math.BigInteger;
+import java.util.concurrent.ThreadLocalRandom;
+
+public class DiffContext {
+
+    public final String keyspace;
+    public final TableSpec table;
+    public final BigInteger startToken;
+    public final BigInteger endToken;
+    public final DiffCluster source;
+    public final DiffCluster target;
+    private final SpecificTokens specificTokens;
+    private final double reverseReadProbability;
+
+    public DiffContext(final DiffCluster source,
+                       final DiffCluster target,
+                       final String keyspace,
+                       final TableSpec table,
+                       final BigInteger startToken,
+                       final BigInteger endToken,
+                       final SpecificTokens specificTokens,
+                       final double reverseReadProbability) {
+        this.keyspace = keyspace;
+        this.table = table;
+        this.startToken = startToken;
+        this.endToken = endToken;
+        this.source = source;
+        this.target = target;
+        this.specificTokens = specificTokens;
+        this.reverseReadProbability = reverseReadProbability;
+    }
+
+    public boolean shouldReverse() {
+        return ! table.getClusteringColumns().isEmpty()
+               && reverseReadProbability > ThreadLocalRandom.current().nextDouble();
+    }
+
+    public boolean isTokenAllowed(BigInteger token) {
+        return specificTokens.test(token);
+    }
+
+    public static String cqlizedString(final String str) {
+        if (str.toLowerCase().equals(str)) {
+            return str;
+        } else {
+            return "\"" + str + "\"";
+        }
+    }
+
+    public String toString() {
+        return String.format("DiffContext: [keyspace: %s, start_token: %s, end_token: %s]",
+                             keyspace, startToken, endToken);
+    }
+}
diff --git a/spark-job/src/main/java/org/apache/cassandra/diff/DiffJob.java b/spark-job/src/main/java/org/apache/cassandra/diff/DiffJob.java
new file mode 100644
index 0000000..bb14c25
--- /dev/null
+++ b/spark-job/src/main/java/org/apache/cassandra/diff/DiffJob.java
@@ -0,0 +1,296 @@
+/*
+ * 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.diff;
+
+import java.io.Serializable;
+import java.math.BigInteger;
+import java.util.*;
+import java.util.stream.Collectors;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.ImmutableList;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.Session;
+import org.apache.spark.SparkConf;
+import org.apache.spark.SparkFiles;
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.sql.SparkSession;
+
+/**
+ * To run this, something like this should be executed for local runs
+ *
+ * spark-submit --files ./spark-job/localconfig.yaml
+ *              --master "local[2]"
+ *              --class org.apache.cassandra.DiffJob spark-job/target/spark-job-0.1-SNAPSHOT.jar
+ *              localconfig.yaml
+ */
+
+public class DiffJob {
+    private static final Logger logger = LoggerFactory.getLogger(DiffJob.class);
+
+    public static void main(String ... args) {
+        if (args.length == 0) {
+            System.exit(-1);
+        }
+        SparkSession spark = SparkSession.builder().appName("cassandra-diff").getOrCreate();
+        JavaSparkContext sc = new JavaSparkContext(spark.sparkContext());
+        String configFile = SparkFiles.get(args[0]);
+        YamlJobConfiguration configuration = YamlJobConfiguration.load(configFile);
+        DiffJob diffJob = new DiffJob();
+        diffJob.run(configuration, sc);
+        spark.stop();
+    }
+
+    public void run(JobConfiguration configuration, JavaSparkContext sc) {
+        SparkConf conf = sc.getConf();
+        // get partitioner from both clusters and verify that they match
+        ClusterProvider sourceProvider = ClusterProvider.getProvider(configuration.clusterConfig("source"), "source");
+        ClusterProvider targetProvider = ClusterProvider.getProvider(configuration.clusterConfig("target"), "target");
+        String sourcePartitioner;
+        String targetPartitioner;
+        try (Cluster sourceCluster = sourceProvider.getCluster();
+             Cluster targetCluster = targetProvider.getCluster()) {
+            sourcePartitioner = sourceCluster.getMetadata().getPartitioner();
+            targetPartitioner = targetCluster.getMetadata().getPartitioner();
+        }
+        if (!sourcePartitioner.equals(targetPartitioner)) {
+            throw new IllegalStateException(String.format("Cluster partitioners do not match; Source: %s, Target: %s,",
+                                                          sourcePartitioner, targetPartitioner));
+        }
+        TokenHelper tokenHelper = TokenHelper.forPartitioner(sourcePartitioner);
+
+        logger.info("Configuring job metadata store");
+        ClusterProvider metadataProvider = ClusterProvider.getProvider(configuration.clusterConfig("metadata"), "metadata");
+        JobMetadataDb.JobLifeCycle job = null;
+        UUID jobId = null;
+        try (Cluster metadataCluster = metadataProvider.getCluster();
+             Session metadataSession = metadataCluster.connect()) {
+
+            MetadataKeyspaceOptions metadataOptions = configuration.metadataOptions();
+            JobMetadataDb.Schema.maybeInitialize(metadataSession, metadataOptions);
+
+            // Job params, which once a job is created cannot be modified in subsequent re-runs
+            logger.info("Creating or retrieving job parameters");
+            job = new JobMetadataDb.JobLifeCycle(metadataSession, metadataOptions.keyspace);
+            Params params = getJobParams(job, configuration);
+            logger.info("Job Params: {}", params);
+            if (null == params)
+                throw new RuntimeException("Unable to initialize job params");
+
+            jobId = params.jobId;
+            List<Split> splits = getSplits(configuration, TokenHelper.forPartitioner(sourcePartitioner));
+
+            // Job options, which may be modified per-run
+            int instances = Integer.parseInt(conf.get("spark.executor.instances", "4"));
+            int cores = Integer.parseInt(conf.get("spark.executor.cores", "2"));
+            int executors = instances * cores;
+            // according to https://spark.apache.org/docs/latest/rdd-programming-guide.html#parallelized-collections we should
+            // have 2-4 partitions per cpu in the cluster:
+            int slices = Math.min(4 * executors, splits.size());
+            int perExecutorRateLimit = configuration.rateLimit() / executors;
+
+            // Record the high level job summary info
+            job.initializeJob(params,
+                              sourceProvider.getClusterName(),
+                              sourceProvider.toString(),
+                              targetProvider.getClusterName(),
+                              targetProvider.toString());
+
+            logger.info("DiffJob {} comparing [{}] in keyspace {} on {} and {}",
+                        jobId,
+                        String.join(",", params.tables),
+                        params.keyspace,
+                        sourceProvider,
+                        targetProvider);
+
+            // Run the distributed diff and collate results
+            Map<String, RangeStats> diffStats = sc.parallelize(splits, slices)
+                                                  .map((split) -> new Differ(configuration,
+                                                                             params,
+                                                                             perExecutorRateLimit,
+                                                                             split,
+                                                                             tokenHelper,
+                                                                             sourceProvider,
+                                                                             targetProvider,
+                                                                             metadataProvider,
+                                                                             new TrackerProvider(configuration.metadataOptions().keyspace))
+                                                                  .run())
+                                                  .reduce(Differ::accumulate);
+            // Publish results. This also removes the job from the currently running list
+            job.finalizeJob(params.jobId, diffStats);
+            logger.info("FINISHED: {}", diffStats);
+        } catch (Exception e) {
+            // If the job errors out, try and mark the job as not running, so it can be restarted.
+            // If the error was thrown from JobMetadataDb.finalizeJob *after* the job had already
+            // been marked not running, this will log a warning, but is not fatal.
+            if (job != null && jobId != null)
+                job.markNotRunning(jobId);
+            throw new RuntimeException("Diff job failed", e);
+        } finally {
+            if (sc.isLocal())
+                Differ.shutdown();
+        }
+    }
+
+    private static Params getJobParams(JobMetadataDb.JobLifeCycle job, JobConfiguration conf) {
+        if (conf.jobId().isPresent()) {
+            return job.getJobParams(conf.jobId().get());
+        } else {
+            return new Params(UUID.randomUUID(),
+                              conf.keyspace(),
+                              conf.tables(),
+                              conf.buckets(),
+                              conf.splits());
+        }
+    }
+
+    private static List<Split> getSplits(JobConfiguration config, TokenHelper tokenHelper) {
+        logger.info("Initializing splits");
+        List<Split> splits = calculateSplits(config.splits(), config.buckets(), tokenHelper);
+        logger.info("All Splits: {}", splits);
+        if (!config.specificTokens().isEmpty() && config.specificTokens().modifier == SpecificTokens.Modifier.ACCEPT) {
+            splits = getSplitsForTokens(config.specificTokens().tokens, splits);
+            logger.info("Splits for specific tokens ONLY: {}", splits);
+        }
+        // shuffle the splits to make sure the work is spread over the workers,
+        // important if it isn't a full cluster is being compared
+        Collections.shuffle(splits);
+        return splits;
+    }
+
+    @VisibleForTesting
+    static List<Split> calculateSplits(int numSplits, int numBuckets, TokenHelper tokenHelper) {
+        List<Split> splits = new ArrayList<>(numSplits);
+        BigInteger minToken = tokenHelper.min();
+        BigInteger maxToken = tokenHelper.max();
+
+        BigInteger totalTokens = maxToken.subtract(minToken);
+        BigInteger segmentSize = totalTokens.divide(BigInteger.valueOf(numSplits));
+
+        // add the first split starting at minToken without adding BigInt.ONE below
+        // Splits are grouped into buckets so we can shard the journal info across
+        // C* partitions
+        splits.add(new Split(0,  0, minToken, minToken.add(segmentSize)));
+        BigInteger prev = minToken.add(segmentSize);
+        for (int i = 1; i < numSplits - 1; i++) {
+            BigInteger next = prev.add(segmentSize);
+            // add ONE to avoid split overlap
+            splits.add(new Split(i, i % numBuckets, prev.add(BigInteger.ONE), next));
+            prev = next;
+        }
+        splits.add(new Split(numSplits - 1, (numSplits - 1) % numBuckets,  prev.add(BigInteger.ONE), maxToken)); // make sure we cover the whole range
+        return splits;
+    }
+
+    @VisibleForTesting
+    static List<Split> getSplitsForTokens(Set<BigInteger> tokens, List<Split> splits) {
+        return splits.stream().filter(split -> split.containsAny(tokens)).collect(Collectors.toList());
+    }
+
+    @VisibleForTesting
+    static class Split implements Serializable {
+        final int splitNumber;
+        final int bucket;
+        final BigInteger start;
+        final BigInteger end;
+
+        Split(int splitNumber, int bucket, BigInteger start, BigInteger end) {
+            this.splitNumber = splitNumber;
+            this.bucket = bucket;
+            this.start = start;
+            this.end = end;
+        }
+
+        public String toString() {
+            return "Split [" +
+                   start +
+                   ", " +
+                   end +
+                   ']';
+        }
+
+        public boolean containsAny(Set<BigInteger> specificTokens) {
+            for (BigInteger specificToken : specificTokens) {
+                if (specificToken.compareTo(start) >= 0 && specificToken.compareTo(end) <= 0)
+                    return true;
+            }
+            return false;
+        }
+    }
+
+    static class Params implements Serializable {
+        public final UUID jobId;
+        public final String keyspace;
+        public final ImmutableList<String> tables;
+        public final int buckets;
+        public final int tasks;
+
+        Params(UUID jobId, String keyspace, List<String> tables, int buckets, int tasks) {
+            this.jobId = jobId;
+            this.keyspace = keyspace;
+            this.tables = ImmutableList.copyOf(tables);
+            this.buckets = buckets;
+            this.tasks = tasks;
+        }
+
+        public String toString() {
+            return String.format("Params: [jobId: %s, keyspace: %s, tables: %s, buckets: %s, tasks: %s]",
+                                 jobId, keyspace, tables.stream().collect(Collectors.joining(",")), buckets, tasks);
+        }
+    }
+
+    static class TaskStatus {
+        public static final TaskStatus EMPTY = new TaskStatus(null, null);
+        public final BigInteger lastToken;
+        public final RangeStats stats;
+
+        TaskStatus(BigInteger lastToken, RangeStats stats) {
+            this.lastToken = lastToken;
+            this.stats = stats;
+        }
+
+        public String toString() {
+            return "TaskStatus{" +
+                   "lastToken=" + lastToken +
+                   ", stats=" + stats +
+                   '}';
+        }
+    }
+
+    public static class TrackerProvider implements Serializable {
+        private final String metadataKeyspace;
+
+        TrackerProvider(String metadataKeyspace) {
+            this.metadataKeyspace = metadataKeyspace;
+        }
+
+        public void initializeStatements(Session session) {
+            JobMetadataDb.ProgressTracker.initializeStatements(session, metadataKeyspace);
+        }
+
+        public JobMetadataDb.ProgressTracker getTracker(Session session, UUID jobId, Split split) {
+            return new JobMetadataDb.ProgressTracker(jobId, split.bucket, split.start, split.end, metadataKeyspace, session);
+        }
+    }
+ }
diff --git a/spark-job/src/main/java/org/apache/cassandra/diff/Differ.java b/spark-job/src/main/java/org/apache/cassandra/diff/Differ.java
new file mode 100644
index 0000000..49576a2
--- /dev/null
+++ b/spark-job/src/main/java/org/apache/cassandra/diff/Differ.java
@@ -0,0 +1,326 @@
+/*
+ * 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.diff;
+
+import java.io.PrintWriter;
+import java.io.Serializable;
+import java.io.StringWriter;
+import java.math.BigInteger;
+import java.util.*;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.function.BiConsumer;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Verify;
+import com.google.common.util.concurrent.*;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.datastax.driver.core.ConsistencyLevel;
+import com.datastax.driver.core.Session;
+
+import com.codahale.metrics.MetricRegistry;
+import com.codahale.metrics.Timer;
+
+public class Differ implements Serializable
+{
+    private static final Logger logger = LoggerFactory.getLogger(Differ.class);
+
+    private static final MetricRegistry metrics = new MetricRegistry();
+
+    private static final int COMPARISON_THREADS = 8;
+    private static final ComparisonExecutor COMPARISON_EXECUTOR = ComparisonExecutor.newExecutor(COMPARISON_THREADS, metrics);
+
+    private final UUID jobId;
+    private final DiffJob.Split split;
+    private final TokenHelper tokenHelper;
+    private final String keyspace;
+    private final List<String> tables;
+    private final RateLimiter rateLimiter;
+    private final DiffJob.TrackerProvider trackerProvider;
+    private final double reverseReadProbability;
+    private final SpecificTokens specificTokens;
+
+    private static DiffCluster srcDiffCluster;
+    private static DiffCluster targetDiffCluster;
+    private static Session journalSession;
+
+    static
+    {
+        Thread.setDefaultUncaughtExceptionHandler((t, e) -> {
+            StringWriter stackTrace = new StringWriter();
+            e.printStackTrace(new PrintWriter(stackTrace));
+            System.out.println("UNCAUGHT EXCEPTION: " + stackTrace.toString());
+            throw new RuntimeException(e);
+        });
+        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+            logger.info("In shutdown hook");
+            shutdown();
+        }));
+    }
+
+    public Differ(JobConfiguration config,
+                  DiffJob.Params params,
+                  int perExecutorRateLimit,
+                  DiffJob.Split split,
+                  TokenHelper tokenHelper,
+                  ClusterProvider sourceProvider,
+                  ClusterProvider targetProvider,
+                  ClusterProvider metadataProvider,
+                  DiffJob.TrackerProvider trackerProvider)
+    {
+        logger.info("Creating Differ for {}", split);
+        this.jobId = params.jobId;
+        this.split = split;
+        this.tokenHelper = tokenHelper;
+        this.keyspace = params.keyspace;
+        this.tables = params.tables;
+        this.trackerProvider = trackerProvider;
+        rateLimiter = RateLimiter.create(perExecutorRateLimit);
+        this.reverseReadProbability = config.reverseReadProbability();
+        this.specificTokens = config.specificTokens();
+        synchronized (Differ.class)
+        {
+            /*
+            Spark runs jobs on each worker in the same JVM, we need to initialize these only once, otherwise
+            we run OOM with health checker threads
+             */
+            // yes we could have JobConfiguration return this directly, but snakeyaml doesn't like relocated classes and the driver has to be shaded
+            ConsistencyLevel cl = ConsistencyLevel.valueOf(config.consistencyLevel());
+            if (srcDiffCluster == null)
+            {
+                srcDiffCluster = new DiffCluster(DiffCluster.Type.SOURCE,
+                                                 sourceProvider.getCluster(),
+                                                 params.keyspace,
+                                                 cl,
+                                                 rateLimiter,
+                                                 config.tokenScanFetchSize(),
+                                                 config.partitionReadFetchSize(),
+                                                 config.readTimeoutMillis());
+            }
+
+            if (targetDiffCluster == null)
+            {
+                targetDiffCluster = new DiffCluster(DiffCluster.Type.TARGET,
+                                                    targetProvider.getCluster(),
+                                                    params.keyspace,
+                                                    cl,
+                                                    rateLimiter,
+                                                    config.tokenScanFetchSize(),
+                                                    config.partitionReadFetchSize(),
+                                                    config.readTimeoutMillis());
+            }
+
+            if (journalSession == null)
+            {
+                journalSession = metadataProvider.getCluster().connect();
+                trackerProvider.initializeStatements(journalSession);
+            }
+        }
+    }
+
+    public Map<String, RangeStats> run() {
+        JobMetadataDb.ProgressTracker journal = trackerProvider.getTracker(journalSession, jobId, split);
+        Map<String, DiffJob.TaskStatus> tablesToDiff = filterTables(tables,
+                                                                    split,
+                                                                    journal::getLastStatus,
+                                                                    !specificTokens.isEmpty());
+
+        String metricsPrefix = String.format("%s.%s", srcDiffCluster.clusterId.name(), srcDiffCluster.keyspace);
+        logger.info("Diffing {} for tables {}", split, tablesToDiff);
+
+        for (Map.Entry<String, DiffJob.TaskStatus> tableStatus : tablesToDiff.entrySet()) {
+            final String table = tableStatus.getKey();
+            DiffJob.TaskStatus status = tableStatus.getValue();
+            RangeStats diffStats = status.stats;
+
+            // if this split has already been fully processed, it's being re-run to check
+            // partitions with errors. In that case, we don't want to adjust the split
+            // start and we don't want to update the completed count when we're finished.
+            boolean isRerun = split.end.equals(status.lastToken);
+            BigInteger startToken = status.lastToken == null || isRerun ? split.start : status.lastToken;
+            validateRange(startToken, split.end, tokenHelper);
+
+            TableSpec sourceTable = TableSpec.make(table, srcDiffCluster);
+            TableSpec targetTable = TableSpec.make(table, targetDiffCluster);
+            validateTableSpecs(sourceTable, targetTable);
+
+            DiffContext ctx = new DiffContext(srcDiffCluster,
+                                              targetDiffCluster,
+                                              keyspace,
+                                              sourceTable,
+                                              startToken,
+                                              split.end,
+                                              specificTokens,
+                                              reverseReadProbability);
+
+            String timerName = String.format("%s.%s.split_times", metricsPrefix, table);
+            try (@SuppressWarnings("unused") Timer.Context timer = metrics.timer(timerName).time()) {
+                diffStats.accumulate(diffTable(ctx,
+                                               (error, token) -> journal.recordError(table, token, error),
+                                               (type, token) -> journal.recordMismatch(table, type, token),
+                                               (stats, token) -> journal.updateStatus(table, stats, token)));
+
+                // update the journal with the final state for the table. Use the split's ending token
+                // as the last seen token (even though we may not have actually read any partition for
+                // that token) as this effectively marks the split as done.
+                journal.finishTable(table, diffStats, !isRerun);
+            }
+        }
+
+        Map<String, RangeStats> statsByTable = tablesToDiff.entrySet()
+                                                           .stream()
+                                                           .collect(Collectors.toMap(Map.Entry::getKey,
+                                                                                     e -> e.getValue().stats));
+        updateMetrics(metricsPrefix, statsByTable);
+        return statsByTable;
+    }
+
+    public RangeStats diffTable(final DiffContext context,
+                                final BiConsumer<Throwable, BigInteger> partitionErrorReporter,
+                                final BiConsumer<MismatchType, BigInteger> mismatchReporter,
+                                final BiConsumer<RangeStats, BigInteger> journal) {
+
+        final Iterator<PartitionKey> sourceKeys = context.source.getPartitionKeys(context.table.getTable(),
+                                                                                  context.startToken,
+                                                                                  context.endToken);
+        final Iterator<PartitionKey> targetKeys = context.target.getPartitionKeys(context.table.getTable(),
+                                                                                  context.startToken,
+                                                                                  context.endToken);
+        final Function<PartitionKey, PartitionComparator> partitionTaskProvider =
+            (key) -> {
+                boolean reverse = context.shouldReverse();
+                return new PartitionComparator(context.table,
+                                               context.source.getPartition(context.table, key, reverse),
+                                               context.target.getPartition(context.table, key, reverse));
+            };
+
+        RangeComparator rangeComparator = new RangeComparator(context,
+                                                              partitionErrorReporter,
+                                                              mismatchReporter,
+                                                              journal,
+                                                              COMPARISON_EXECUTOR);
+
+        final RangeStats tableStats = rangeComparator.compare(sourceKeys, targetKeys, partitionTaskProvider);
+        logger.debug("Table [{}] stats - ({})", context.table.getTable(), tableStats);
+        return tableStats;
+    }
+
+    @VisibleForTesting
+    static Map<String, DiffJob.TaskStatus> filterTables(Iterable<String> tables,
+                                                        DiffJob.Split split,
+                                                        Function<String, DiffJob.TaskStatus> journal,
+                                                        boolean includeCompleted) {
+        Map<String, DiffJob.TaskStatus> tablesToProcess = new HashMap<>();
+        for (String table : tables) {
+            DiffJob.TaskStatus taskStatus = journal.apply(table);
+            RangeStats diffStats = taskStatus.stats;
+            BigInteger lastToken = taskStatus.lastToken;
+
+            // When we finish processing a split for a given table, we update the task status in journal
+            // to set the last seen token to the split's end token, to indicate that the split is complete.
+            if (!includeCompleted && lastToken != null && lastToken.equals(split.end)) {
+                logger.info("Found finished table {} for split {}", table, split);
+            }
+            else {
+                tablesToProcess.put(table, diffStats != null
+                                            ? taskStatus
+                                            : new DiffJob.TaskStatus(taskStatus.lastToken, RangeStats.newStats()));
+            }
+        }
+        return tablesToProcess;
+    }
+
+    static void validateTableSpecs(TableSpec source, TableSpec target) {
+        Verify.verify(source.equalsNamesOnly(target),
+                      "Source and target table definitions do not match (Source: %s Target: %s)",
+                      source, target);
+    }
+
+    @VisibleForTesting
+    static void validateRange(BigInteger start, BigInteger end, TokenHelper tokens) {
+
+        Verify.verify(start != null && end != null, "Invalid token range [%s,%s]", start, end);
+
+        Verify.verify(start.compareTo(tokens.min()) >= 0 && end.compareTo(tokens.max()) <= 0 && start.compareTo(end) < 0,
+                      "Invalid token range [%s,%s] for partitioner range [%s,%s]",
+                       start, end, tokens.min(), tokens.max());
+    }
+
+    @VisibleForTesting
+    static Map<String, RangeStats> accumulate(Map<String, RangeStats> stats, Map<String, RangeStats> otherStats)
+    {
+        for (Map.Entry<String, RangeStats> otherEntry : otherStats.entrySet())
+        {
+            if (stats.containsKey(otherEntry.getKey()))
+                stats.get(otherEntry.getKey()).accumulate(otherEntry.getValue());
+            else
+                stats.put(otherEntry.getKey(), otherEntry.getValue());
+        }
+        return stats;
+    }
+
+    private static void updateMetrics(String prefix, Map<String, RangeStats> statsMap)
+    {
+        for (Map.Entry<String, RangeStats> entry : statsMap.entrySet())
+        {
+            String qualifier = String.format("%s.%s", prefix, entry.getKey());
+            RangeStats stats = entry.getValue();
+
+            metrics.meter(qualifier + ".partitions_read").mark(stats.getMatchedPartitions() + stats.getOnlyInSource() + stats.getOnlyInTarget() + stats.getMismatchedPartitions());
+            metrics.counter(qualifier + ".matched_partitions").inc(stats.getMatchedPartitions());
+            metrics.counter(qualifier + ".mismatched_partitions").inc(stats.getMismatchedPartitions());
+
+            metrics.counter(qualifier + ".partitions_only_in_source").inc(stats.getOnlyInSource());
+            metrics.counter(qualifier + ".partitions_only_in_target").inc(stats.getOnlyInTarget());
+            metrics.counter(qualifier + ".skipped_partitions").inc(stats.getSkippedPartitions());
+
+            metrics.counter(qualifier + ".matched_rows").inc(stats.getMatchedRows());
+            metrics.counter(qualifier + ".matched_values").inc(stats.getMatchedValues());
+            metrics.counter(qualifier + ".mismatched_values").inc(stats.getMismatchedValues());
+        }
+    }
+
+    public static void shutdown()
+    {
+        try
+        {
+            if (srcDiffCluster != null) {
+                srcDiffCluster.stop();
+                srcDiffCluster.close();
+            }
+            if (targetDiffCluster != null) {
+                targetDiffCluster.stop();
+                targetDiffCluster.close();
+            }
+            if (journalSession != null) {
+                journalSession.close();
+                journalSession.getCluster().close();
+            }
+            COMPARISON_EXECUTOR.shutdown();
+        }
+        catch (Exception e)
+        {
+            throw new RuntimeException(e);
+        }
+    }
+}
diff --git a/spark-job/src/main/java/org/apache/cassandra/diff/JobMetadataDb.java b/spark-job/src/main/java/org/apache/cassandra/diff/JobMetadataDb.java
new file mode 100644
index 0000000..0ac6521
--- /dev/null
+++ b/spark-job/src/main/java/org/apache/cassandra/diff/JobMetadataDb.java
@@ -0,0 +1,567 @@
+/*
+ * 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.diff;
+
+import java.math.BigInteger;
+import java.util.Map;
+import java.util.UUID;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.datastax.driver.core.*;
+import com.datastax.driver.core.utils.UUIDs;
+import org.joda.time.DateTime;
+import org.joda.time.DateTimeZone;
+
+public class JobMetadataDb {
+    private static final Logger logger = LoggerFactory.getLogger(JobMetadataDb.class);
+
+    static class ProgressTracker {
+
+        private final UUID jobId;
+        private final int bucket;
+        private final String startToken;
+        private final String endToken;
+        private final String keyspace;
+        private Session session;
+
+        private static PreparedStatement updateStmt;
+        private static PreparedStatement mismatchStmt;
+        private static PreparedStatement errorSummaryStmt;
+        private static PreparedStatement errorDetailStmt;
+        private static PreparedStatement updateCompleteStmt;
+
+        public ProgressTracker(UUID jobId,
+                               int bucket,
+                               BigInteger startToken,
+                               BigInteger endToken,
+                               String keyspace,
+                               Session session) {
+            this.jobId = jobId;
+            this.bucket = bucket;
+            this.startToken = startToken.toString();
+            this.endToken = endToken.toString();
+            this.keyspace = keyspace;
+            this.session = session;
+        }
+
+        /**
+         * Runs on each executor to prepare statements shared across all instances
+         */
+        public static void initializeStatements(Session session, String keyspace) {
+            if (updateStmt == null) {
+                updateStmt = session.prepare(String.format("INSERT INTO %s.%s (" +
+                                                           " job_id," +
+                                                           " bucket," +
+                                                           " table_name," +
+                                                           " start_token," +
+                                                           " end_token," +
+                                                           " matched_partitions," +
+                                                           " mismatched_partitions," +
+                                                           " partitions_only_in_source," +
+                                                           " partitions_only_in_target," +
+                                                           " matched_rows," +
+                                                           " matched_values," +
+                                                           " mismatched_values," +
+                                                           " skipped_partitions," +
+                                                           " last_token )" +
+                                                           "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+                                                           keyspace, Schema.TASK_STATUS));
+            }
+            if (mismatchStmt == null) {
+                mismatchStmt = session.prepare(String.format("INSERT INTO %s.%s (" +
+                                                             " job_id," +
+                                                             " bucket," +
+                                                             " table_name," +
+                                                             " mismatching_token," +
+                                                             " mismatch_type )" +
+                                                             "VALUES (?, ?, ?, ?, ?)",
+                                                             keyspace, Schema.MISMATCHES));
+            }
+            if (updateCompleteStmt == null) {
+                updateCompleteStmt = session.prepare(String.format("UPDATE %s.%s " +
+                                                                   " SET completed = completed + 1" +
+                                                                   " WHERE job_id = ? " +
+                                                                   " AND bucket = ? " +
+                                                                   " AND table_name = ? ",
+                                                                   keyspace, Schema.JOB_STATUS))
+                                            .setConsistencyLevel(ConsistencyLevel.LOCAL_QUORUM);
+            }
+            if (errorSummaryStmt == null) {
+                errorSummaryStmt = session.prepare(String.format("INSERT INTO %s.%s (" +
+                                                                 " job_id," +
+                                                                 " bucket," +
+                                                                 " table_name," +
+                                                                 " start_token," +
+                                                                 " end_token)" +
+                                                                 " VALUES (?, ?, ?, ?, ?)",
+                                                                 keyspace, Schema.ERROR_SUMMARY));
+            }
+            if (errorDetailStmt == null) {
+                errorDetailStmt = session.prepare(String.format("INSERT INTO %s.%s (" +
+                                                                " job_id," +
+                                                                " bucket," +
+                                                                " table_name," +
+                                                                " start_token," +
+                                                                " end_token," +
+                                                                " error_token)" +
+                                                                " VALUES (?, ?, ?, ?, ?, ?)",
+                                                                keyspace, Schema.ERROR_DETAIL));
+            }
+
+        }
+
+        /**
+         *
+         * @param table
+         * @return
+         */
+        public DiffJob.TaskStatus getLastStatus(String table) {
+            ResultSet rs = session.execute(String.format("SELECT last_token, " +
+                                                         "       matched_partitions, " +
+                                                         "       mismatched_partitions, " +
+                                                         "       partitions_only_in_source, " +
+                                                         "       partitions_only_in_target, " +
+                                                         "       matched_rows," +
+                                                         "       matched_values," +
+                                                         "       mismatched_values," +
+                                                         "       skipped_partitions " +
+                                                         " FROM %s.%s " +
+                                                         " WHERE job_id = ? " +
+                                                         " AND   bucket = ? " +
+                                                         " AND   table_name = ? " +
+                                                         " AND   start_token = ? " +
+                                                         " AND   end_token = ?",
+                                                         keyspace, Schema.TASK_STATUS),
+                                           jobId, bucket, table, startToken, endToken);
+            Row row = rs.one();
+            if (null == row)
+                return DiffJob.TaskStatus.EMPTY;
+
+            RangeStats stats = RangeStats.withValues(getOrDefaultLong(row, "matched_partitions"),
+                                                     getOrDefaultLong(row, "mismatched_partitions"),
+                                                     0L, // error counts are per-run and not persisted in the metadata db
+                                                     getOrDefaultLong(row, "skipped_partitions"),
+                                                     getOrDefaultLong(row, "partitions_only_in_source"),
+                                                     getOrDefaultLong(row, "partitions_only_in_target"),
+                                                     getOrDefaultLong(row, "matched_rows"),
+                                                     getOrDefaultLong(row, "matched_values"),
+                                                     getOrDefaultLong(row, "mismatched_values"));
+
+            BigInteger lastToken = row.isNull("last_token") ? null : new BigInteger(row.getString("last_token"));
+            return new DiffJob.TaskStatus(lastToken, stats);
+        }
+
+        /**
+         *
+         * @param table
+         * @param diffStats
+         * @param latestToken
+         */
+        public void updateStatus(String table, RangeStats diffStats, BigInteger latestToken) {
+            session.execute(bindUpdateStatement(table, diffStats, latestToken));
+        }
+
+        public void recordMismatch(String table, MismatchType type, BigInteger token) {
+            logger.info("Detected mismatch in table {}; partition with token {} is {}",
+                        table, token, type == MismatchType.PARTITION_MISMATCH
+                                      ? " different in source and target clusters"
+                                      : type == MismatchType.ONLY_IN_SOURCE ? "only present in source cluster"
+                                                                            : "only present in target cluster");
+            session.execute(bindMismatchesStatement(table, token, type.name()));
+        }
+
+        /**
+         *
+         * @param table
+         * @param token
+         * @param error
+         */
+        public void recordError(String table, BigInteger token, Throwable error) {
+            logger.error(String.format("Encountered error during partition comparison in table %s; " +
+                                       "error for partition with token %s", table, token), error);
+            BatchStatement batch = new BatchStatement();
+            batch.add(bindErrorSummaryStatement(table));
+            batch.add(bindErrorDetailStatement(table, token));
+            batch.setIdempotent(true);
+            session.execute(batch);
+        }
+
+        /**
+         *
+         * @param table
+         * @param stats
+         */
+        public void finishTable(String table, RangeStats stats, boolean updateCompletedCount) {
+            logger.info("Finishing range [{}, {}] for table {}", startToken, endToken, table);
+            // first flush out the last status.
+            session.execute(bindUpdateStatement(table, stats, endToken));
+            // then update the count of completed tasks
+            if (updateCompletedCount)
+                session.execute(updateCompleteStmt.bind(jobId, bucket, table));
+        }
+
+        private Statement bindMismatchesStatement(String table, BigInteger token, String type) {
+            return mismatchStmt.bind(jobId, bucket, table, token.toString(), type)
+                               .setIdempotent(true);
+        }
+
+        private Statement bindErrorSummaryStatement(String table) {
+            return errorSummaryStmt.bind(jobId, bucket, table, startToken, endToken)
+                                   .setIdempotent(true);
+        }
+
+        private Statement bindErrorDetailStatement(String table, BigInteger errorToken) {
+            return errorDetailStmt.bind(jobId, bucket, table, startToken, endToken, errorToken.toString())
+                                  .setIdempotent(true);
+        }
+
+        private Statement bindUpdateStatement(String table, RangeStats stats, BigInteger token) {
+           return bindUpdateStatement(table, stats, token.toString());
+        }
+
+        private Statement bindUpdateStatement(String table, RangeStats stats, String token) {
+            // We don't persist the partition error count from RangeStats as errors
+            // are likely to be transient and not data related, so we don't want to
+            // accumulate them across runs.
+            return updateStmt.bind(jobId,
+                                   bucket,
+                                   table,
+                                   startToken,
+                                   endToken,
+                                   stats.getMatchedPartitions(),
+                                   stats.getMismatchedPartitions(),
+                                   stats.getOnlyInSource(),
+                                   stats.getOnlyInTarget(),
+                                   stats.getMatchedRows(),
+                                   stats.getMatchedValues(),
+                                   stats.getMismatchedValues(),
+                                   stats.getSkippedPartitions(),
+                                   token)
+                             .setIdempotent(true);
+        }
+
+        private static long getOrDefaultLong(Row row, String column) {
+            return (null == row || row.isNull(column)) ? 0L : row.getLong(column);
+        }
+    }
+
+    static class JobLifeCycle {
+        final Session session;
+        final String keyspace;
+
+        public JobLifeCycle(Session session, String metadataKeyspace) {
+            this.session = session;
+            this.keyspace = metadataKeyspace;
+        }
+
+        public DiffJob.Params getJobParams(UUID jobId) {
+            ResultSet rs = session.execute(String.format("SELECT keyspace_name, " +
+                                                         "       table_names," +
+                                                         "       buckets," +
+                                                         "       total_tasks " +
+                                                         "FROM %s.%s " +
+                                                         "WHERE job_id = ?",
+                                                         keyspace, Schema.JOB_SUMMARY),
+                                           jobId);
+            Row row = rs.one();
+            if (null == row)
+                return null;
+
+            return new DiffJob.Params(jobId,
+                                      row.getString("keyspace_name"),
+                                      row.getList("table_names", String.class),
+                                      row.getInt("buckets"),
+                                      row.getInt("total_tasks"));
+        }
+
+
+        // Runs on Driver to insert top level job info
+        public void initializeJob(DiffJob.Params params,
+                                  String sourceClusterName,
+                                  String sourceClusterDesc,
+                                  String targetClusterName,
+                                  String targetClusterDesc) {
+
+            logger.info("Initializing job status");
+            // The job was previously run, so this could be a re-run to
+            // mop up any failed splits so mark it in progress.
+            ResultSet rs = session.execute(String.format("INSERT INTO %s.%s (job_id) VALUES (?) IF NOT EXISTS",
+                                                         keyspace, Schema.RUNNING_JOBS),
+                                           params.jobId);
+            if (!rs.one().getBool("[applied]")) {
+                logger.info("Aborting due to inability to mark job as running. " +
+                            "Did a previous run of job id {} fail non-gracefully?",
+                            params.jobId);
+                throw new RuntimeException("Unable to mark job running, aborting");
+            }
+
+            UUID timeUUID = UUIDs.timeBased();
+            DateTime startDateTime = new DateTime(UUIDs.unixTimestamp(timeUUID), DateTimeZone.UTC);
+
+            rs = session.execute(String.format("INSERT INTO %s.%s (" +
+                                               " job_id," +
+                                               " job_start_time," +
+                                               " buckets," +
+                                               " keyspace_name," +
+                                               " table_names," +
+                                               " source_cluster_name," +
+                                               " source_cluster_desc," +
+                                               " target_cluster_name," +
+                                               " target_cluster_desc," +
+                                               " total_tasks)" +
+                                               " VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)" +
+                                               " IF NOT EXISTS",
+                                               keyspace, Schema.JOB_SUMMARY),
+                                 params.jobId,
+                                 timeUUID,
+                                 params.buckets,
+                                 params.keyspace,
+                                 params.tables,
+                                 sourceClusterName,
+                                 sourceClusterDesc,
+                                 targetClusterName,
+                                 targetClusterDesc,
+                                 params.tasks);
+
+            // This is a brand new job, index its details including start time
+            if (rs.one().getBool("[applied]")) {
+                BatchStatement batch = new BatchStatement();
+                batch.add(new SimpleStatement(String.format("INSERT INTO %s.%s (source_cluster_name, job_id) VALUES (?, ?)",
+                                                            keyspace, Schema.SOURCE_CLUSTER_INDEX),
+                                              sourceClusterName, params.jobId));
+                batch.add(new SimpleStatement(String.format("INSERT INTO %s.%s (target_cluster_name, job_id) VALUES (?, ?)",
+                                                            keyspace, Schema.TARGET_CLUSTER_INDEX),
+                                              targetClusterName, params.jobId));
+                batch.add(new SimpleStatement(String.format("INSERT INTO %s.%s (keyspace_name, job_id) VALUES (?, ?)",
+                                                            keyspace, Schema.KEYSPACE_INDEX),
+                                              keyspace, params.jobId));
+                batch.add(new SimpleStatement(String.format("INSERT INTO %s.%s (job_start_date, job_start_hour, job_start_time, job_id) " +
+                                                            "VALUES ('%s', ?, ?, ?)",
+                                                            keyspace, Schema.JOB_START_INDEX, startDateTime.toString("yyyy-MM-dd")),
+                                              startDateTime.getHourOfDay(), timeUUID, params.jobId));
+                session.execute(batch);
+            }
+        }
+
+        public void finalizeJob(UUID jobId, Map<String, RangeStats> results) {
+            logger.info("Finalizing job status");
+
+            markNotRunning(jobId);
+
+            BatchStatement batch = new BatchStatement();
+            for (Map.Entry<String, RangeStats> result : results.entrySet()) {
+                String table = result.getKey();
+                RangeStats stats = result.getValue();
+                session.execute(String.format("INSERT INTO %s.%s (" +
+                                              "  job_id," +
+                                              "  table_name," +
+                                              "  matched_partitions," +
+                                              "  mismatched_partitions," +
+                                              "  partitions_only_in_source," +
+                                              "  partitions_only_in_target," +
+                                              "  matched_rows," +
+                                              "  matched_values," +
+                                              "  mismatched_values," +
+                                              "  skipped_partitions) " +
+                                              "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+                                              keyspace, Schema.JOB_RESULTS),
+                                jobId,
+                                table,
+                                stats.getMatchedPartitions(),
+                                stats.getMismatchedPartitions(),
+                                stats.getOnlyInSource(),
+                                stats.getOnlyInTarget(),
+                                stats.getMatchedRows(),
+                                stats.getMatchedValues(),
+                                stats.getMismatchedValues(),
+                                stats.getSkippedPartitions());
+            }
+            session.execute(batch);
+        }
+
+
+        public void markNotRunning(UUID jobId) {
+            logger.info("Marking job {} as not running", jobId);
+
+            ResultSet rs = session.execute(String.format("DELETE FROM %s.%s WHERE job_id = ? IF EXISTS",
+                                                         keyspace, Schema.RUNNING_JOBS),
+                                           jobId);
+            if (!rs.one().getBool("[applied]")) {
+                logger.warn("Non-fatal: Unable to mark job %s as not running, check logs for errors " +
+                            "during initialization as there may be no entry for this job in the {} table",
+                            jobId, Schema.RUNNING_JOBS);
+            }
+        }
+    }
+
+    static class Schema {
+
+        public static final String TASK_STATUS = "task_status";
+        private static final String TASK_STATUS_SCHEMA = "CREATE TABLE IF NOT EXISTS %s.%s (" +
+                                                         " job_id uuid," +
+                                                         " bucket int," +
+                                                         " table_name text," +
+                                                         " start_token varchar," +
+                                                         " end_token varchar," +
+                                                         " matched_partitions bigint," +
+                                                         " mismatched_partitions bigint, " +
+                                                         " partitions_only_in_source bigint," +
+                                                         " partitions_only_in_target bigint," +
+                                                         " matched_rows bigint," +
+                                                         " matched_values bigint," +
+                                                         " mismatched_values bigint," +
+                                                         " skipped_partitions bigint," +
+                                                         " last_token varchar," +
+                                                         " PRIMARY KEY((job_id, bucket), table_name, start_token, end_token))" +
+                                                         " WITH default_time_to_live = %s";
+
+        public static final String JOB_SUMMARY = "job_summary";
+        private static final String JOB_SUMMARY_SCHEMA = "CREATE TABLE IF NOT EXISTS %s.%s (" +
+                                                         " job_id uuid," +
+                                                         " job_start_time timeuuid," +
+                                                         " buckets int," +
+                                                         " keyspace_name text," +
+                                                         " table_names frozen<list<text>>," +
+                                                         " source_cluster_name text," +
+                                                         " source_cluster_desc text," +
+                                                         " target_cluster_name text," +
+                                                         " target_cluster_desc text," +
+                                                         " total_tasks int," +
+                                                         " PRIMARY KEY(job_id))" +
+                                                         " WITH default_time_to_live = %s";
+
+        public static final String JOB_RESULTS = "job_results";
+        private static final String JOB_RESULTS_SCHEMA = "CREATE TABLE IF NOT EXISTS %s.%s (" +
+                                                         " job_id uuid," +
+                                                         " table_name text," +
+                                                         " matched_partitions bigint," +
+                                                         " mismatched_partitions bigint," +
+                                                         " partitions_only_in_source bigint," +
+                                                         " partitions_only_in_target bigint," +
+                                                         " matched_rows bigint," +
+                                                         " matched_values bigint," +
+                                                         " mismatched_values bigint," +
+                                                         " skipped_partitions bigint," +
+                                                         " PRIMARY KEY(job_id, table_name))" +
+                                                         " WITH default_time_to_live = %s";
+
+        public static final String JOB_STATUS = "job_status";
+        private static final String JOB_STATUS_SCHEMA = "CREATE TABLE IF NOT EXISTS %s.%s (" +
+                                                        " job_id uuid," +
+                                                        " bucket int," +
+                                                        " table_name text," +
+                                                        " completed counter," +
+                                                        " PRIMARY KEY ((job_id, bucket), table_name))";
+
+        public static final String MISMATCHES = "mismatches";
+        private static final String MISMATCHES_SCHEMA = "CREATE TABLE IF NOT EXISTS %s.%s (" +
+                                                        " job_id uuid," +
+                                                        " bucket int," +
+                                                        " table_name text, " +
+                                                        " mismatching_token varchar, " +
+                                                        " mismatch_type text, " +
+                                                        " PRIMARY KEY ((job_id, bucket), table_name, mismatching_token))" +
+                                                        " WITH default_time_to_live = %s";
+
+        public static final String ERROR_SUMMARY = "task_errors";
+        private static final String ERROR_SUMMARY_SCHEMA = "CREATE TABLE IF NOT EXISTS %s.%s (" +
+                                                           " job_id uuid," +
+                                                           " bucket int," +
+                                                           " table_name text," +
+                                                           " start_token varchar," +
+                                                           " end_token varchar," +
+                                                           " PRIMARY KEY ((job_id, bucket), table_name, start_token, end_token))" +
+                                                           " WITH default_time_to_live = %s";
+
+        public static final String ERROR_DETAIL = "partition_errors";
+        private static final String ERROR_DETAIL_SCHEMA = "CREATE TABLE IF NOT EXISTS %s.%s (" +
+                                                          " job_id uuid," +
+                                                          " bucket int," +
+                                                          " table_name text," +
+                                                          " start_token varchar," +
+                                                          " end_token varchar," +
+                                                          " error_token varchar," +
+                                                          " PRIMARY KEY ((job_id, bucket, table_name, start_token, end_token), error_token))" +
+                                                          " WITH default_time_to_live = %s";
+
+        public static final String SOURCE_CLUSTER_INDEX = "source_cluster_index";
+        private static final String SOURCE_CLUSTER_INDEX_SCHEMA = "CREATE TABLE IF NOT EXISTS %s.%s (" +
+                                                                  " source_cluster_name text," +
+                                                                  " job_id uuid," +
+                                                                  " PRIMARY KEY (source_cluster_name, job_id))" +
+                                                                  " WITH default_time_to_live = %s";
+
+        public static final String TARGET_CLUSTER_INDEX = "target_cluster_index";
+        private static final String TARGET_CLUSTER_INDEX_SCHEMA = "CREATE TABLE IF NOT EXISTS %s.%s (" +
+                                                                  " target_cluster_name text," +
+                                                                  " job_id uuid," +
+                                                                  " PRIMARY KEY (target_cluster_name, job_id))" +
+                                                                  " WITH default_time_to_live = %s";
+
+        public static final String KEYSPACE_INDEX = "keyspace_index";
+        private static final String KEYSPACE_INDEX_SCHEMA = "CREATE TABLE IF NOT EXISTS %s.%s (" +
+                                                            " keyspace_name text," +
+                                                            " job_id uuid," +
+                                                            " PRIMARY KEY(keyspace_name, job_id))" +
+                                                            " WITH default_time_to_live = %s";
+
+        public static final String JOB_START_INDEX = "job_start_index";
+        private static final String JOB_START_INDEX_SCHEMA = "CREATE TABLE IF NOT EXISTS %s.%s (" +
+                                                             " job_start_date date," +
+                                                             " job_start_hour int," +
+                                                             " job_start_time timeuuid," +
+                                                             " job_id uuid," +
+                                                             " PRIMARY KEY ((job_start_date, job_start_hour), job_start_time))" +
+                                                             " WITH default_time_to_live = %s";
+
+        public static final String RUNNING_JOBS = "running_jobs";
+        private static final String RUNNING_JOBS_SCHEMA = "CREATE TABLE IF NOT EXISTS %s.%s (" +
+                                                          " job_id uuid," +
+                                                          " PRIMARY KEY (job_id))" +
+                                                          " WITH default_time_to_live = %s";
+
+        private static final String KEYSPACE_SCHEMA = "CREATE KEYSPACE IF NOT EXISTS %s WITH REPLICATION = %s";
+
+
+        public static void maybeInitialize(Session session, MetadataKeyspaceOptions options) {
+            if (!options.should_init)
+                return;
+
+            logger.info("Initializing cassandradiff journal schema in \"{}\" keyspace", options.keyspace);
+            session.execute(String.format(KEYSPACE_SCHEMA, options.keyspace, options.replication));
+            session.execute(String.format(JOB_SUMMARY_SCHEMA, options.keyspace, JOB_SUMMARY, options.ttl));
+            session.execute(String.format(JOB_STATUS_SCHEMA, options.keyspace, JOB_STATUS));
+            session.execute(String.format(JOB_RESULTS_SCHEMA, options.keyspace, JOB_RESULTS, options.ttl));
+            session.execute(String.format(TASK_STATUS_SCHEMA, options.keyspace, TASK_STATUS, options.ttl));
+            session.execute(String.format(MISMATCHES_SCHEMA, options.keyspace, MISMATCHES, options.ttl));
+            session.execute(String.format(ERROR_SUMMARY_SCHEMA, options.keyspace, ERROR_SUMMARY, options.ttl));
+            session.execute(String.format(ERROR_DETAIL_SCHEMA, options.keyspace, ERROR_DETAIL, options.ttl));
+            session.execute(String.format(SOURCE_CLUSTER_INDEX_SCHEMA, options.keyspace, SOURCE_CLUSTER_INDEX, options.ttl));
+            session.execute(String.format(TARGET_CLUSTER_INDEX_SCHEMA, options.keyspace, TARGET_CLUSTER_INDEX, options.ttl));
+            session.execute(String.format(KEYSPACE_INDEX_SCHEMA, options.keyspace, KEYSPACE_INDEX, options.ttl));
+            session.execute(String.format(JOB_START_INDEX_SCHEMA, options.keyspace, JOB_START_INDEX, options.ttl));
+            session.execute(String.format(RUNNING_JOBS_SCHEMA, options.keyspace, RUNNING_JOBS, options.ttl));
+            logger.info("Schema initialized");
+        }
+    }
+}
+
diff --git a/spark-job/src/main/java/org/apache/cassandra/diff/MismatchType.java b/spark-job/src/main/java/org/apache/cassandra/diff/MismatchType.java
new file mode 100644
index 0000000..b23ac35
--- /dev/null
+++ b/spark-job/src/main/java/org/apache/cassandra/diff/MismatchType.java
@@ -0,0 +1,28 @@
+/*
+ * 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.diff;
+
+public enum MismatchType {
+
+    ONLY_IN_SOURCE,
+    ONLY_IN_TARGET,
+    PARTITION_MISMATCH
+
+}
diff --git a/spark-job/src/main/java/org/apache/cassandra/diff/PartitionComparator.java b/spark-job/src/main/java/org/apache/cassandra/diff/PartitionComparator.java
new file mode 100644
index 0000000..6434dc8
--- /dev/null
+++ b/spark-job/src/main/java/org/apache/cassandra/diff/PartitionComparator.java
@@ -0,0 +1,114 @@
+/*
+ * 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.diff;
+
+import java.util.Iterator;
+import java.util.concurrent.*;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.datastax.driver.core.*;
+
+public class PartitionComparator implements Callable<PartitionStats> {
+
+    private static final Logger logger = LoggerFactory.getLogger(PartitionComparator.class);
+
+    private final TableSpec tableSpec;
+    private final Iterator<Row> source;
+    private final Iterator<Row> target;
+
+    public PartitionComparator(TableSpec tableSpec,
+                               Iterator<Row> source,
+                               Iterator<Row> target) {
+        this.tableSpec = tableSpec;
+        this.source = source;
+        this.target = target;
+    }
+
+    public PartitionStats call() {
+        PartitionStats partitionStats = new PartitionStats();
+
+        if (source == null || target == null) {
+            logger.error("Skipping partition because one result was null (timeout despite retries)");
+            partitionStats.skipped = true;
+            return partitionStats;
+        }
+
+        while (source.hasNext() && target.hasNext()) {
+
+            Row sourceRow = source.next();
+            Row targetRow = target.next();
+
+            // if primary keys don't match don't proceed any further, just mark the
+            // partition as mismatched and be done
+            if (!clusteringsEqual(sourceRow, targetRow)) {
+                partitionStats.allClusteringsMatch = false;
+                return partitionStats;
+            }
+
+            partitionStats.matchedRows++;
+
+            // if the rows match, but there are mismatching values in the regular columns
+            // we can continue processing the partition, so just flag it as mismatched and continue
+            checkRegularColumnEquality(partitionStats, sourceRow, targetRow);
+        }
+
+        // if one of the iterators isn't exhausted, then there's a mismatch at the partition level
+        if (source.hasNext() || target.hasNext())
+            partitionStats.allClusteringsMatch = false;
+
+        return partitionStats;
+    }
+
+    private boolean clusteringsEqual(Row source, Row target) {
+        for (ColumnMetadata column : tableSpec.getClusteringColumns()) {
+            Object fromSource = source.getObject(column.getName());
+            Object fromTarget = target.getObject(column.getName());
+
+            if ((fromSource == null) != (fromTarget == null))
+                return false;
+
+            if (fromSource != null && !fromSource.equals(fromTarget))
+                return false;
+        }
+        return true;
+    }
+
+    private void checkRegularColumnEquality(PartitionStats stats, Row source, Row target) {
+        for (ColumnMetadata column : tableSpec.getRegularColumns()) {
+            Object fromSource = source.getObject(column.getName());
+            Object fromTarget = target.getObject(column.getName());
+            if (fromSource == null) {
+                if (fromTarget == null) {
+                    stats.matchedValues++;
+                } else {
+                    stats.mismatchedValues++;
+                }
+            } else {
+                if (fromSource.equals(fromTarget)) {
+                    stats.matchedValues++;
+                } else {
+                    stats.mismatchedValues++;
+                }
+            }
+        }
+    }
+}
diff --git a/spark-job/src/main/java/org/apache/cassandra/diff/PartitionKey.java b/spark-job/src/main/java/org/apache/cassandra/diff/PartitionKey.java
new file mode 100644
index 0000000..d31da4f
--- /dev/null
+++ b/spark-job/src/main/java/org/apache/cassandra/diff/PartitionKey.java
@@ -0,0 +1,85 @@
+/*
+ * 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.diff;
+
+import java.math.BigInteger;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import com.datastax.driver.core.*;
+import org.jetbrains.annotations.NotNull;
+
+public class PartitionKey implements Comparable<PartitionKey> {
+
+    private final Row row;
+
+    public PartitionKey(Row row) {
+        this.row = row;
+    }
+
+    public BigInteger getTokenAsBigInteger(){
+        Token token = getToken();
+        if (token.getType() == DataType.bigint()) {
+            return BigInteger.valueOf((Long) token.getValue());
+        } else {
+            return (BigInteger) token.getValue();
+        }
+    }
+
+    public List<Object> getComponents() {
+        int cols = row.getColumnDefinitions().size();
+        List<Object> columns = new ArrayList<>(cols);
+        // Note we start at index=1, because index=0 is the token
+        for (int i = 1; i < cols; i++)
+            columns.add(row.getObject(i));
+        return columns;
+    }
+
+    @VisibleForTesting
+    protected Token getToken() {
+        return row.getToken(0);
+    }
+
+    public int compareTo(@NotNull PartitionKey o) {
+        return getToken().compareTo(o.getToken());
+    }
+
+    public boolean equals(Object obj) {
+        return this == obj || (obj instanceof PartitionKey &&  this.compareTo((PartitionKey)obj) == 0);
+    }
+
+    public int hashCode() {
+        return Objects.hash(getTokenAsBigInteger());
+    }
+
+    public String toString() {
+        return StreamSupport.stream(row.getColumnDefinitions().spliterator(), false)
+                            .map(ColumnDefinitions.Definition::getName)
+                            .map(row::getObject)
+                            .filter(Objects::nonNull)
+                            .map(Object::toString)
+                            .collect(Collectors.joining(":"));
+    }
+}
diff --git a/spark-job/src/main/java/org/apache/cassandra/diff/PartitionStats.java b/spark-job/src/main/java/org/apache/cassandra/diff/PartitionStats.java
new file mode 100644
index 0000000..7f020ff
--- /dev/null
+++ b/spark-job/src/main/java/org/apache/cassandra/diff/PartitionStats.java
@@ -0,0 +1,28 @@
+/*
+ * 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.diff;
+
+public class PartitionStats {
+    public boolean skipped = false;
+    public boolean allClusteringsMatch = true;
+    public long matchedRows;
+    public long matchedValues;
+    public long mismatchedValues;
+}
diff --git a/spark-job/src/main/java/org/apache/cassandra/diff/RangeComparator.java b/spark-job/src/main/java/org/apache/cassandra/diff/RangeComparator.java
new file mode 100644
index 0000000..36ce2b5
--- /dev/null
+++ b/spark-job/src/main/java/org/apache/cassandra/diff/RangeComparator.java
@@ -0,0 +1,232 @@
+/*
+ * 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.diff;
+
+import java.math.BigInteger;
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.*;
+
+import com.google.common.base.Verify;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class RangeComparator {
+
+    private static final Logger logger = LoggerFactory.getLogger(RangeComparator.class);
+
+    private final DiffContext context;
+    private final BiConsumer<Throwable, BigInteger> errorReporter;
+    private final BiConsumer<MismatchType, BigInteger> mismatchReporter;
+    private final BiConsumer<RangeStats, BigInteger> journal;
+    private final ComparisonExecutor comparisonExecutor;
+
+    public RangeComparator(DiffContext context,
+                           BiConsumer<Throwable, BigInteger> errorReporter,
+                           BiConsumer<MismatchType,BigInteger> mismatchReporter,
+                           BiConsumer<RangeStats, BigInteger> journal,
+                           ComparisonExecutor comparisonExecutor) {
+        this.context = context;
+        this.errorReporter = errorReporter;
+        this.mismatchReporter = mismatchReporter;
+        this.journal = journal;
+        this.comparisonExecutor = comparisonExecutor;
+    }
+
+    public RangeStats compare(Iterator<PartitionKey> sourceKeys,
+                              Iterator<PartitionKey> targetKeys,
+                              Function<PartitionKey, PartitionComparator> partitionTaskProvider) {
+
+        final RangeStats rangeStats = RangeStats.newStats();
+        // We can catch this condition earlier, but it doesn't hurt to also check here
+        if (context.startToken.equals(context.endToken))
+            return rangeStats;
+
+        Phaser phaser = new Phaser(1);
+        AtomicLong partitionCount = new AtomicLong(0);
+        AtomicReference<BigInteger> highestTokenSeen = new AtomicReference<>(context.startToken);
+
+        logger.info("Comparing range [{},{}]", context.startToken, context.endToken);
+        try {
+            PartitionKey sourceKey = nextKey(sourceKeys);
+            PartitionKey targetKey = nextKey(targetKeys);
+
+            // special case for start of range - handles one cluster supplying an empty range
+            if ((sourceKey == null) != (targetKey == null)) {
+                if (sourceKey == null) {
+                    logger.info("First in range, source iter is empty {}", context);
+                    onlyInTarget(rangeStats, targetKey);
+                    targetKeys.forEachRemaining(key -> onlyInTarget(rangeStats, key));
+                } else {
+                    logger.info("First in range, target iter is empty {}", context);
+                    onlyInSource(rangeStats, sourceKey);
+                    sourceKeys.forEachRemaining(key -> onlyInSource(rangeStats, key));
+                }
+                return rangeStats;
+            }
+
+            while (sourceKey != null && targetKey != null) {
+
+                int ret = sourceKey.compareTo(targetKey);
+                if (ret > 0) {
+                    onlyInTarget(rangeStats, targetKey);
+                    targetKey = nextKey(targetKeys);
+                } else if (ret < 0) {
+                    onlyInSource(rangeStats, sourceKey);
+                    sourceKey = nextKey(sourceKeys);
+                } else {
+
+                    Verify.verify(sourceKey.equals(targetKey),
+                                  "Can only compare partitions with identical keys: (%s, %s)",
+                                  sourceKey, targetKey);
+
+                    // For results where the key exists in both, we'll fire off an async task to walk the
+                    // partition and compare all the rows. The result of that comparison is added to the
+                    // totals for the range and the highest seen token updated in the onSuccess callback
+
+                    if (!context.isTokenAllowed(sourceKey.getTokenAsBigInteger())) {
+                        logger.debug("Skipping disallowed token {}", sourceKey.getTokenAsBigInteger());
+                        rangeStats.skipPartition();
+                        sourceKey = nextKey(sourceKeys);
+                        targetKey = nextKey(targetKeys);
+                        continue;
+                    }
+
+                    BigInteger token = sourceKey.getTokenAsBigInteger();
+                    try {
+                        PartitionComparator comparisonTask = partitionTaskProvider.apply(sourceKey);
+                        comparisonExecutor.submit(comparisonTask,
+                                                  onSuccess(rangeStats, partitionCount, token, highestTokenSeen, mismatchReporter, journal),
+                                                  onError(rangeStats, token, errorReporter),
+                                                  phaser);
+                    } catch (Throwable t) {
+                        // Handle errors thrown when creating the comparison task. This should trap timeouts and
+                        // unavailables occurring when performing the initial query to read the full partition.
+                        // Errors thrown when paging through the partition in comparisonTask will be handled by
+                        // the onError callback.
+                        rangeStats.partitionError();
+                        errorReporter.accept(t, token);
+                    } finally {
+                        // if the cluster has been shutdown because the task failed the underlying iterators
+                        // of partition keys will return hasNext == false
+                        sourceKey = nextKey(sourceKeys);
+                        targetKey = nextKey(targetKeys);
+                    }
+                }
+            }
+
+            // handle case where only one iterator is exhausted
+            if (sourceKey != null)
+                onlyInSource(rangeStats, sourceKey);
+            else if (targetKey != null)
+                onlyInTarget(rangeStats, targetKey);
+
+            drain(sourceKeys, targetKeys, rangeStats);
+
+        } catch (Exception e) {
+            // Handles errors thrown by iteration of underlying resultsets of partition keys by
+            // calls to nextKey(). Such errors should cause the overall range comparison to fail,
+            // but we must ensure that any in-flight partition comparisons complete so that either
+            // the onSuccess or onError callback is fired for each one. This is necessary to ensure
+            // that we record the highest seen token and any failed partitions and can safely re-run.
+            logger.debug("Waiting for {} in flight tasks before propagating error", phaser.getUnarrivedParties());
+            phaser.arriveAndAwaitAdvance();
+            throw new RuntimeException(String.format("Error encountered during range comparison for [%s:%s]",
+                                       context.startToken, context.endToken), e);
+        }
+
+        logger.debug("Waiting for {} in flight tasks before returning", phaser.getUnarrivedParties());
+        phaser.arriveAndAwaitAdvance();
+
+        if (!rangeStats.allMatches())
+            logger.info("Segment [{}:{}] stats - ({})", context.startToken, context.endToken, rangeStats);
+
+        return rangeStats;
+    }
+
+    private void drain(Iterator<PartitionKey> sourceKeys,
+                             Iterator<PartitionKey> targetKeys,
+                             RangeStats rangeStats) {
+        if (sourceKeys.hasNext()) {
+            logger.info("Source keys not exhausted {}", context);
+            sourceKeys.forEachRemaining(key -> onlyInSource(rangeStats, key));
+        } else if (targetKeys.hasNext()) {
+            logger.info("Target keys not exhausted: {}", context);
+            targetKeys.forEachRemaining(key -> onlyInTarget(rangeStats, key));
+        }
+    }
+
+    private void onlyInTarget(RangeStats stats, PartitionKey key) {
+        stats.onlyInTarget();
+        mismatchReporter.accept(MismatchType.ONLY_IN_TARGET, key.getTokenAsBigInteger());
+    }
+
+    private void onlyInSource(RangeStats stats, PartitionKey key) {
+        stats.onlyInSource();
+        mismatchReporter.accept(MismatchType.ONLY_IN_SOURCE, key.getTokenAsBigInteger());
+    }
+
+    private PartitionKey nextKey(Iterator<PartitionKey> keys) {
+        return keys.hasNext() ? keys.next() : null;
+    }
+
+    private Consumer<PartitionStats> onSuccess(final RangeStats rangeStats,
+                                               final AtomicLong partitionCount,
+                                               final BigInteger currentToken,
+                                               final AtomicReference<BigInteger> highestSeenToken,
+                                               final BiConsumer<MismatchType, BigInteger> mismatchReporter,
+                                               final BiConsumer<RangeStats, BigInteger> journal) {
+        return (result) -> {
+
+            rangeStats.accumulate(result);
+            if (!result.allClusteringsMatch || result.mismatchedValues > 0) {
+                mismatchReporter.accept(MismatchType.PARTITION_MISMATCH, currentToken);
+                rangeStats.mismatchedPartition();
+            } else {
+                rangeStats.matchedPartition();
+            }
+
+            BigInteger highest = highestSeenToken.get();
+            while (currentToken.compareTo(highest) > 0) {
+                if (highestSeenToken.compareAndSet(highest, currentToken))
+                    break;
+
+                highest = highestSeenToken.get();
+            }
+
+            // checkpoint ever 10 partitions
+            if (partitionCount.incrementAndGet() % 10 == 0)
+                journal.accept(rangeStats, highestSeenToken.get());
+        };
+    }
+
+    private Consumer<Throwable> onError(final RangeStats rangeStats,
+                                        final BigInteger currentToken,
+                                        final BiConsumer<Throwable, BigInteger> errorReporter) {
+        return (error) -> {
+            rangeStats.partitionError();
+            errorReporter.accept(error, currentToken);
+        };
+    }
+}
+
+
diff --git a/spark-job/src/main/java/org/apache/cassandra/diff/RangeStats.java b/spark-job/src/main/java/org/apache/cassandra/diff/RangeStats.java
new file mode 100644
index 0000000..a0f8043
--- /dev/null
+++ b/spark-job/src/main/java/org/apache/cassandra/diff/RangeStats.java
@@ -0,0 +1,286 @@
+/*
+ * 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.diff;
+
+import java.io.*;
+import java.util.Objects;
+import java.util.concurrent.atomic.LongAdder;
+
+public class RangeStats implements Serializable {
+
+    private transient LongAdder matchedPartitions;
+    private transient LongAdder mismatchedPartitions;
+    private transient LongAdder errorPartitions;
+    private transient LongAdder skippedPartitions;
+    private transient LongAdder onlyInSource;
+    private transient LongAdder onlyInTarget;
+    private transient LongAdder matchedRows;
+    private transient LongAdder matchedValues;
+    private transient LongAdder mismatchedValues;
+
+    public static RangeStats newStats() {
+        return new RangeStats(0L, 0L, 0L, 0L, 0L, 0L, 0L, 0L, 0L);
+    }
+
+    public static RangeStats withValues(long matchedPartitions,
+                                        long mismatchedPartitions,
+                                        long errorPartitions,
+                                        long skippedPartitions,
+                                        long onlyInSource,
+                                        long onlyInTarget,
+                                        long matchedRows,
+                                        long matchedValues,
+                                        long mismatchedValues) {
+
+        return new RangeStats(matchedPartitions,
+                              mismatchedPartitions,
+                              errorPartitions,
+                              skippedPartitions,
+                              onlyInSource,
+                              onlyInTarget,
+                              matchedRows,
+                              matchedValues,
+                              mismatchedValues);
+    }
+
+    private RangeStats(long matchedPartitions,
+                       long mismatchedPartitions,
+                       long errorPartitions,
+                       long skippedPartitions,
+                       long onlyInSource,
+                       long onlyInTarget,
+                       long matchedRows,
+                       long matchedValues,
+                       long mismatchedValues) {
+
+        this.matchedPartitions      = new LongAdder();
+        this.mismatchedPartitions   = new LongAdder();
+        this.errorPartitions        = new LongAdder();
+        this.skippedPartitions      = new LongAdder();
+        this.onlyInSource           = new LongAdder();
+        this.onlyInTarget           = new LongAdder();
+        this.matchedRows            = new LongAdder();
+        this.matchedValues          = new LongAdder();
+        this.mismatchedValues       = new LongAdder();
+
+        this.matchedPartitions.add(matchedPartitions);
+        this.mismatchedPartitions.add(mismatchedPartitions);
+        this.errorPartitions.add(errorPartitions);
+        this.skippedPartitions.add(skippedPartitions);
+        this.onlyInSource.add(onlyInSource);
+        this.onlyInTarget.add(onlyInTarget);
+        this.matchedRows.add(matchedRows);
+        this.matchedValues.add(matchedValues);
+        this.mismatchedValues.add(mismatchedValues);
+    }
+
+    public void matchedPartition() {
+        matchedPartitions.add(1L);
+    }
+
+    public long getMatchedPartitions() {
+        return matchedPartitions.sum();
+    }
+
+    public void onlyInSource() {
+        onlyInSource.add(1L);
+    }
+
+    public long getOnlyInSource() {
+        return onlyInSource.sum();
+    }
+
+    public void onlyInTarget() {
+        onlyInTarget.add(1L);
+    }
+
+    public long getOnlyInTarget() {
+        return onlyInTarget.sum();
+    }
+
+    public long getMatchedRows() {
+        return matchedRows.sum();
+    }
+
+    public long getMatchedValues() {
+        return matchedValues.sum();
+    }
+
+    public long getMismatchedValues() {
+        return mismatchedValues.sum();
+    }
+
+    public void mismatchedPartition() {
+        mismatchedPartitions.add(1L);
+    }
+
+    public long getMismatchedPartitions() {
+        return mismatchedPartitions.sum();
+    }
+
+    public void skipPartition() {
+        skippedPartitions.add(1L);
+    }
+
+    public long getSkippedPartitions() {
+        return skippedPartitions.sum();
+    }
+
+    public void partitionError() {
+        errorPartitions.add(1L);
+    }
+
+    public long getErrorPartitions() {
+        return errorPartitions.sum();
+    }
+
+    public RangeStats accumulate(final PartitionStats partitionStats) {
+        this.matchedRows.add(partitionStats.matchedRows);
+        this.matchedValues.add(partitionStats.matchedValues);
+        this.mismatchedValues.add(partitionStats.mismatchedValues);
+        if (partitionStats.skipped)
+            this.skippedPartitions.add(1L);
+
+        return this;
+    }
+
+    public RangeStats accumulate(final RangeStats rangeStats) {
+        this.matchedPartitions.add(rangeStats.matchedPartitions.sum());
+        this.mismatchedPartitions.add(rangeStats.mismatchedPartitions.sum());
+        this.errorPartitions.add(rangeStats.errorPartitions.sum());
+        this.skippedPartitions.add(rangeStats.skippedPartitions.sum());
+        this.onlyInSource.add(rangeStats.onlyInSource.sum());
+        this.onlyInTarget.add(rangeStats.onlyInTarget.sum());
+        this.matchedRows.add(rangeStats.matchedRows.sum());
+        this.matchedValues.add(rangeStats.matchedValues.sum());
+        this.mismatchedValues.add(rangeStats.mismatchedValues.sum());
+        return this;
+    }
+
+    public boolean allMatches () {
+        return onlyInSource.sum() == 0
+               && errorPartitions.sum() == 0
+               && onlyInTarget.sum() == 0
+               && mismatchedValues.sum() == 0
+               && skippedPartitions.sum() == 0;
+    }
+
+    public boolean isEmpty() {
+        return matchedPartitions.sum() == 0
+                && mismatchedPartitions.sum() == 0
+                && errorPartitions.sum() == 0
+                && skippedPartitions.sum() == 0
+                && onlyInSource.sum() == 0
+                && onlyInTarget.sum() == 0
+                && matchedRows.sum() == 0
+                && matchedValues.sum() == 0
+                && mismatchedValues.sum() == 0;
+    }
+
+    public boolean equals(Object o) {
+        if (this == o)
+            return true;
+
+        if (!(o instanceof RangeStats))
+            return false;
+
+        RangeStats other = (RangeStats)o;
+        return this.matchedPartitions.sum() == other.matchedPartitions.sum()
+               && mismatchedPartitions.sum() == other.mismatchedPartitions.sum()
+               && errorPartitions.sum() == other.errorPartitions.sum()
+               && skippedPartitions.sum() == other.skippedPartitions.sum()
+               && onlyInSource.sum() == other.onlyInSource.sum()
+               && onlyInTarget.sum() == other.onlyInTarget.sum()
+               && matchedRows.sum() == other.matchedRows.sum()
+               && matchedValues.sum() == other.matchedValues.sum()
+               && mismatchedValues.sum() == other.mismatchedValues.sum();
+    }
+
+    public int hashCode() {
+        return Objects.hash(matchedPartitions.sum(),
+                            mismatchedPartitions.sum(),
+                            errorPartitions.sum(),
+                            skippedPartitions.sum(),
+                            onlyInSource.sum(),
+                            onlyInTarget.sum(),
+                            matchedRows.sum(),
+                            matchedValues.sum(),
+                            mismatchedValues.sum());
+    }
+
+    public String toString() {
+        return String.format("Matched Partitions - %d, " +
+                             "Mismatched Partitions - %d, " +
+                             "Partition Errors - %d, " +
+                             "Partitions Only In Source - %d, " +
+                             "Partitions Only In Target - %d, " +
+                             "Skipped Partitions - %d, " +
+                             "Matched Rows - %d, " +
+                             "Matched Values - %d, " +
+                             "Mismatched Values - %d ",
+                             matchedPartitions.sum(),
+                             mismatchedPartitions.sum(),
+                             errorPartitions.sum(),
+                             onlyInSource.sum(),
+                             onlyInTarget.sum(),
+                             skippedPartitions.sum(),
+                             matchedRows.sum(),
+                             matchedValues.sum(),
+                             mismatchedValues.sum());
+    }
+
+    // For serialization
+
+    private RangeStats() {}
+
+    private void writeObject(ObjectOutputStream out) throws IOException {
+        out.writeLong(matchedPartitions.sum());
+        out.writeLong(mismatchedPartitions.sum());
+        out.writeLong(errorPartitions.sum());
+        out.writeLong(skippedPartitions.sum());
+        out.writeLong(onlyInSource.sum());
+        out.writeLong(onlyInTarget.sum());
+        out.writeLong(matchedRows.sum());
+        out.writeLong(matchedValues.sum());
+        out.writeLong(mismatchedValues.sum());
+    }
+
+    private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
+        this.matchedPartitions      = new LongAdder();
+        this.mismatchedPartitions   = new LongAdder();
+        this.errorPartitions        = new LongAdder();
+        this.skippedPartitions      = new LongAdder();
+        this.onlyInSource           = new LongAdder();
+        this.onlyInTarget           = new LongAdder();
+        this.matchedRows            = new LongAdder();
+        this.matchedValues          = new LongAdder();
+        this.mismatchedValues       = new LongAdder();
+
+        this.matchedPartitions.add(in.readLong());
+        this.mismatchedPartitions.add(in.readLong());
+        this.errorPartitions.add(in.readLong());
+        this.skippedPartitions.add(in.readLong());
+        this.onlyInSource.add(in.readLong());
+        this.onlyInTarget.add(in.readLong());
+        this.matchedRows.add(in.readLong());
+        this.matchedValues.add(in.readLong());
+        this.mismatchedValues.add(in.readLong());
+    }
+}
diff --git a/spark-job/src/main/java/org/apache/cassandra/diff/TableSpec.java b/spark-job/src/main/java/org/apache/cassandra/diff/TableSpec.java
new file mode 100644
index 0000000..d2f0963
--- /dev/null
+++ b/spark-job/src/main/java/org/apache/cassandra/diff/TableSpec.java
@@ -0,0 +1,121 @@
+/*
+ * 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.diff;
+
+import com.datastax.driver.core.*;
+import java.util.*;
+import java.util.stream.Collectors;
+
+import com.google.common.base.MoreObjects;
+import com.google.common.collect.ImmutableList;
+
+import static org.apache.cassandra.diff.DiffContext.cqlizedString;
+
+public class TableSpec {
+
+    private final String table;
+    private ImmutableList<ColumnMetadata> clusteringColumns;
+    private ImmutableList<ColumnMetadata> regularColumns;
+
+
+    public String getTable()
+    {
+        return table;
+    }
+
+
+    public ImmutableList<ColumnMetadata> getClusteringColumns() {
+        return clusteringColumns;
+    }
+
+    public ImmutableList<ColumnMetadata> getRegularColumns() {
+        return regularColumns;
+    }
+
+
+    /**
+     * @param table the table to diff
+     * @param clusteringColumns the clustering columns, retrieved from cluster using the client
+     * @param regularColumns the non-primary key columns, retrieved from cluster using the client
+     */
+    TableSpec(final String table,
+              final List<ColumnMetadata> clusteringColumns,
+              final List<ColumnMetadata> regularColumns) {
+        this.table = table;
+        this.clusteringColumns = ImmutableList.copyOf(clusteringColumns);
+        this.regularColumns = ImmutableList.copyOf(regularColumns);
+    }
+
+    public static TableSpec make(String table, DiffCluster diffCluster) {
+        final Cluster cluster = diffCluster.cluster;
+
+        final String cqlizedKeyspace = cqlizedString(diffCluster.keyspace);
+        final String cqlizedTable = cqlizedString(table);
+
+        KeyspaceMetadata ksMetadata = cluster.getMetadata().getKeyspace(cqlizedKeyspace);
+        if (ksMetadata == null) {
+            throw new IllegalArgumentException(String.format("Keyspace %s not found in %s cluster", diffCluster.keyspace, diffCluster.clusterId));
+        }
+
+        TableMetadata tableMetadata = ksMetadata.getTable(cqlizedTable);
+        List<ColumnMetadata> clusteringColumns = tableMetadata.getClusteringColumns();
+        List<ColumnMetadata> regularColumns = tableMetadata.getColumns()
+                                                           .stream()
+                                                           .filter(c -> !(clusteringColumns.contains(c)))
+                                                           .collect(Collectors.toList());
+        return new TableSpec(tableMetadata.getName(), clusteringColumns, regularColumns);
+    }
+
+    public boolean equalsNamesOnly(TableSpec other) {
+        return this.table.equals(other.table)
+            && columnNames(this.clusteringColumns).equals(columnNames(other.clusteringColumns))
+            && columnNames(this.regularColumns).equals(columnNames(other.regularColumns));
+    }
+
+    private static List<String> columnNames(List<ColumnMetadata> columns) {
+        return columns.stream().map(ColumnMetadata::getName).collect(Collectors.toList());
+    }
+
+    public boolean equals(Object o) {
+        if (this == o)
+            return true;
+
+        if (!(o instanceof TableSpec))
+            return false;
+
+        TableSpec other = (TableSpec)o;
+        return this.table.equals(other.table)
+               && this.clusteringColumns.equals(other.clusteringColumns)
+               && this.regularColumns.equals(other.regularColumns);
+
+    }
+
+    public int hashCode() {
+        return Objects.hash(table, clusteringColumns, regularColumns);
+    }
+
+    public String toString() {
+        return MoreObjects.toStringHelper(this)
+                          .add("table", table)
+                          .add("clusteringColumns", clusteringColumns)
+                          .add("regularColumns", regularColumns)
+                          .toString();
+    }
+}
diff --git a/spark-job/src/main/java/org/apache/cassandra/diff/TokenHelper.java b/spark-job/src/main/java/org/apache/cassandra/diff/TokenHelper.java
new file mode 100644
index 0000000..3521a4b
--- /dev/null
+++ b/spark-job/src/main/java/org/apache/cassandra/diff/TokenHelper.java
@@ -0,0 +1,69 @@
+/*
+ * 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.diff;
+
+import java.math.BigInteger;
+
+public enum TokenHelper {
+
+    MURMUR3 {
+        @Override
+        public BigInteger min() {
+            return BigInteger.valueOf(Long.MIN_VALUE);
+        }
+
+        @Override
+        public BigInteger max() {
+            return BigInteger.valueOf(Long.MAX_VALUE);
+        }
+
+        @Override
+        public Object forBindParam(BigInteger token) {
+            return token.longValue();
+        }
+    },
+
+    RANDOM {
+        @Override
+        public BigInteger min() {
+            return BigInteger.ONE.negate();
+        }
+
+        @Override
+        public BigInteger max() {
+            return BigInteger.valueOf(2).pow(127).subtract(BigInteger.ONE);
+        }
+
+        @Override
+        public Object forBindParam(BigInteger token) {
+            return token;
+        }
+    };
+
+    public abstract BigInteger min();
+    public abstract BigInteger max();
+    public abstract Object forBindParam(BigInteger token);
+
+    public static TokenHelper forPartitioner(String partitionerName) {
+        if (partitionerName.endsWith("Murmur3Partitioner")) return MURMUR3;
+        else if (partitionerName.endsWith("RandomPartitioner")) return RANDOM;
+        else throw new IllegalArgumentException("Unsupported Partitioner :" + partitionerName);
+    }
+}
diff --git a/spark-job/src/main/resources/log4j2.xml b/spark-job/src/main/resources/log4j2.xml
new file mode 100644
index 0000000..64aec61
--- /dev/null
+++ b/spark-job/src/main/resources/log4j2.xml
@@ -0,0 +1,38 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ 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.
+  -->
+
+<Configuration status="WARN">
+    <Appenders>
+        <Console name="Console" target="SYSTEM_OUT">
+            <PatternLayout pattern="%d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"/>
+        </Console>
+    </Appenders>
+    <Loggers>
+        <Root level="INFO">
+            <AppenderRef ref="Console"/>
+        </Root>
+        <!--Logger name="org.apache.cassandra.diff" level="info">
+            <AppenderRef ref="Console"/>
+        </Logger>
+        <Logger name="org.apache.cassandra.diff" level="info">
+            <AppenderRef ref="Console"/>
+        </Logger-->
+    </Loggers>
+</Configuration>
\ No newline at end of file
diff --git a/spark-job/src/test/java/com/datastax/driver/core/ColumnMetadataHelper.java b/spark-job/src/test/java/com/datastax/driver/core/ColumnMetadataHelper.java
new file mode 100644
index 0000000..cc7fb95
--- /dev/null
+++ b/spark-job/src/test/java/com/datastax/driver/core/ColumnMetadataHelper.java
@@ -0,0 +1,27 @@
+/*
+ * 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 com.datastax.driver.core;
+
+public class ColumnMetadataHelper {
+
+    public static ColumnMetadata column(String name) {
+        return ColumnMetadata.forAlias(null, name, null);
+    }
+}
diff --git a/spark-job/src/test/java/org/apache/cassandra/diff/ComparisonExecutorTest.java b/spark-job/src/test/java/org/apache/cassandra/diff/ComparisonExecutorTest.java
new file mode 100644
index 0000000..002769f
--- /dev/null
+++ b/spark-job/src/test/java/org/apache/cassandra/diff/ComparisonExecutorTest.java
@@ -0,0 +1,277 @@
+/*
+ * 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.diff;
+
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Consumer;
+
+import com.google.common.util.concurrent.*;
+import org.junit.Test;
+
+import com.codahale.metrics.*;
+
+import static org.apache.cassandra.diff.TestUtils.assertThreadWaits;
+import static junit.framework.TestCase.fail;
+import static org.junit.Assert.assertEquals;
+
+public class ComparisonExecutorTest {
+
+    @Test
+    public void submitBlocksWhenMaxTasksExceeded() throws Exception {
+        // submit maxTasks, then assert that further submission blocks until tasks are processed
+        int maxTasks = 3;
+        MetricRegistry metrics = metrics();
+        final ComparisonExecutor executor = new ComparisonExecutor(executor(1), maxTasks, metrics);
+        Gauge waitingToSubmit = metrics.getGauges().get("BlockedTasks");
+        assertEquals(0, waitingToSubmit.getValue());
+
+        final AtomicInteger successful = new AtomicInteger(0);
+        final Consumer<Integer> onSuccess = (i) -> successful.incrementAndGet();
+        final AtomicInteger failed = new AtomicInteger(0);
+        final Consumer<Throwable> onError = (t) -> failed.incrementAndGet();
+        final Phaser phaser = new Phaser(1);
+
+        BlockingTask[] tasks = new BlockingTask[5];
+        for (int i=0; i<5; i++)
+            tasks[i] = new BlockingTask(i);
+
+        // Ensure that the submission itself does not block before the max number of tasks are submitted
+        executor.submit(tasks[0], onSuccess, onError, phaser);
+        executor.submit(tasks[1], onSuccess, onError, phaser);
+        executor.submit(tasks[2], onSuccess, onError, phaser);
+        assertEquals(0, waitingToSubmit.getValue());
+
+        // Now submit another pair of tasks which should block as the executor is fully occupied
+        final CountDownLatch latch = new CountDownLatch(2);
+        Thread t1 = new Thread(() -> { latch.countDown(); executor.submit(tasks[3], onSuccess, onError, phaser);});
+        Thread t2 = new Thread(() -> { latch.countDown(); executor.submit(tasks[4], onSuccess, onError, phaser);});
+        t1.start();
+        t2.start();
+        // wait for both to attempt submission
+        latch.await();
+        assertThreadWaits(t1);
+        assertThreadWaits(t2);
+        assertEquals(2, waitingToSubmit.getValue());
+
+        // Let the first waiting task complete, which should allow t1 to complete its submission
+        tasks[0].latch.countDown();
+        t1.join();
+        // the second submission should still be waiting on a slot
+        assertThreadWaits(t2);
+        assertEquals(1, waitingToSubmit.getValue());
+
+        // Let another task complete, allowing t2 to complete its submission
+        tasks[1].latch.countDown();
+        t2.join();
+        assertEquals(0, waitingToSubmit.getValue());
+
+        // Let all tasks complete, wait for them to do so then verify counters
+        for (int i=2; i<=4; i++)
+            tasks[i].latch.countDown();
+
+        phaser.arriveAndAwaitAdvance();
+        assertEquals(5, successful.get());
+        assertEquals(0, failed.get());
+    }
+
+    @Test
+    public void handleTaskFailure() {
+        // Ensure that the failure callback is fired, a permit for task submission
+        // returned and the phaser notified when a task throws
+        int maxTasks = 5;
+        MetricRegistry metrics = metrics();
+        ComparisonExecutor executor = new ComparisonExecutor(executor(1), maxTasks, metrics);
+        Gauge availableSlots = metrics.getGauges().get("AvailableSlots");
+        final AtomicInteger successful = new AtomicInteger(0);
+        final Consumer<Integer> onSuccess = (i) -> successful.incrementAndGet();
+        AtomicReference<Throwable> thrown = new AtomicReference<>();
+        final Consumer<Throwable> onError = thrown::set;
+        final Phaser phaser = new Phaser(1);
+
+        assertEquals(maxTasks, availableSlots.getValue());
+
+        RuntimeException toThrow = new RuntimeException("FAIL");
+        BlockingTask task = new BlockingTask(0, toThrow);
+        executor.submit(task, onSuccess, onError, phaser);
+        assertEquals(maxTasks - 1, availableSlots.getValue());
+
+        assertEquals(2, phaser.getUnarrivedParties());
+        task.latch.countDown();
+
+        phaser.arriveAndAwaitAdvance();
+        assertEquals(maxTasks, availableSlots.getValue());
+        assertEquals(0, successful.get());
+        assertEquals(toThrow, thrown.get());
+    }
+
+    @Test
+    public void handleUncaughtExceptionInFailureCallback() {
+        // Ensure that if the failure callback throws, a permit for submission is
+        // still returned and the phaser notified
+        int maxTasks = 5;
+        MetricRegistry metrics = metrics();
+        ComparisonExecutor executor = new ComparisonExecutor(executor(1), maxTasks, metrics);
+        Gauge availableSlots = metrics.getGauges().get("AvailableSlots");
+        final AtomicInteger successful = new AtomicInteger(0);
+        final AtomicInteger failures = new AtomicInteger(0);
+        final Consumer<Integer> onSuccess = (i) -> successful.incrementAndGet();
+        final Consumer<Throwable> onError =  (t) -> { failures.incrementAndGet(); throw new RuntimeException("UNCAUGHT"); };
+        final Phaser phaser = new Phaser(1);
+
+        assertEquals(maxTasks, availableSlots.getValue());
+
+        RuntimeException toThrow = new RuntimeException("FAIL");
+        try {
+            onError.accept(toThrow);
+            fail("Failure callback should throw RuntimeException");
+        } catch (RuntimeException e) {
+            // expected - reset failure count
+            failures.set(0);
+        }
+
+        BlockingTask task = new BlockingTask(0, toThrow);
+        executor.submit(task, onSuccess, onError, phaser);
+        assertEquals(maxTasks - 1, availableSlots.getValue());
+
+        assertEquals(2, phaser.getUnarrivedParties());
+        task.latch.countDown();
+
+        phaser.arriveAndAwaitAdvance();
+        assertEquals(maxTasks, availableSlots.getValue());
+        assertEquals(0, successful.get());
+        assertEquals(1, failures.get());
+    }
+
+    @Test
+    public void handleUncaughtExceptionInSuccessCallback() {
+        // Ensure that if the success callback throws, a permit for submission is
+        // still returned and the phaser notified
+        int maxTasks = 5;
+        MetricRegistry metrics = metrics();
+        ComparisonExecutor executor = new ComparisonExecutor(executor(1), maxTasks, metrics);
+        Gauge availableSlots = metrics.getGauges().get("AvailableSlots");
+        final AtomicInteger successful = new AtomicInteger(0);
+        final AtomicInteger failures = new AtomicInteger(0);
+        final Consumer<Integer> onSuccess = (i) ->  { successful.incrementAndGet(); throw new RuntimeException("UNCAUGHT"); };
+        final Consumer<Throwable> onError =  (t) -> failures.incrementAndGet();
+        final Phaser phaser = new Phaser(1);
+
+        assertEquals(maxTasks, availableSlots.getValue());
+        try {
+            onSuccess.accept(0);
+            fail("Success callback should throw RuntimeException");
+        } catch (RuntimeException e) {
+            // expected - reset failure count
+            successful.set(0);
+        }
+
+        BlockingTask task = new BlockingTask(0);
+        executor.submit(task, onSuccess, onError, phaser);
+        assertEquals(maxTasks - 1, availableSlots.getValue());
+
+        assertEquals(2, phaser.getUnarrivedParties());
+        task.latch.countDown();
+
+        phaser.arriveAndAwaitAdvance();
+        assertEquals(maxTasks, availableSlots.getValue());
+        assertEquals(1, successful.get());
+        assertEquals(0, failures.get());
+    }
+
+    @Test
+    public void handleRejectedExecutionException() {
+        // In the case that the underlying ExecutorService rejects a task submission, a permit
+        // should be returned and the phaser notified
+        int maxTasks = 5;
+        final AtomicInteger successful = new AtomicInteger(0);
+        final AtomicInteger failures = new AtomicInteger(0);
+        final AtomicInteger rejections = new AtomicInteger(0);
+        final Consumer<Integer> onSuccess = (i) ->  successful.incrementAndGet();
+        final Consumer<Throwable> onError =  (t) -> failures.incrementAndGet();
+
+        MetricRegistry metrics = metrics();
+        ExecutorService rejectingExecutor = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,
+                                                                   new LinkedBlockingQueue<>(1),
+                                                                   (r, executor) -> { rejections.incrementAndGet();
+                                                                                      throw new RejectedExecutionException("REJECTED");});
+
+        ComparisonExecutor executor = new ComparisonExecutor(MoreExecutors.listeningDecorator(rejectingExecutor), maxTasks, metrics);
+        Gauge availableSlots = metrics.getGauges().get("AvailableSlots");
+        final Phaser phaser = new Phaser(1);
+
+        // Submit an initial pair of tasks to ensure that the underlying work queue is full
+        BlockingTask t0 = new BlockingTask(0);
+        BlockingTask t1 = new BlockingTask(1);
+        executor.submit(t0, onSuccess, onError, phaser);
+        executor.submit(t1, onSuccess, onError, phaser);
+        assertEquals(3, phaser.getUnarrivedParties());
+        assertEquals(maxTasks - 2, availableSlots.getValue());
+
+        // Submit a third task which will be rejected by the executor service
+        executor.submit(new BlockingTask(2), onSuccess, onError, phaser);
+        t0.latch.countDown();
+        t1.latch.countDown();
+
+        phaser.arriveAndAwaitAdvance();
+        assertEquals(maxTasks, availableSlots.getValue());
+        assertEquals(2, successful.get());
+        assertEquals(1, failures.get());
+        assertEquals(1, rejections.get());
+    }
+
+    class BlockingTask implements Callable<Integer> {
+
+        final int id;
+        final Exception e;
+        final CountDownLatch latch;
+
+        BlockingTask(int id) {
+            this(id, null);
+        }
+
+        BlockingTask(int id, Exception toThrow) {
+            this.id = id;
+            this.e = toThrow;
+            this.latch = new CountDownLatch(1);
+        }
+
+        public Integer call() throws Exception {
+            latch.await();
+            if (e != null)
+                throw e;
+            return id;
+        }
+    }
+
+    private static ListeningExecutorService executor(int threads) {
+        return MoreExecutors.listeningDecorator(
+            Executors.newFixedThreadPool(threads,
+                                         new ThreadFactoryBuilder().setNameFormat("partition-comparison-%d")
+                                                                   .setDaemon(true)
+                                                                   .build()));
+    }
+
+    private static MetricRegistry metrics() {
+        return new MetricRegistry();
+    }
+}
+
diff --git a/spark-job/src/test/java/org/apache/cassandra/diff/DiffJobTest.java b/spark-job/src/test/java/org/apache/cassandra/diff/DiffJobTest.java
new file mode 100644
index 0000000..d8d92c7
--- /dev/null
+++ b/spark-job/src/test/java/org/apache/cassandra/diff/DiffJobTest.java
@@ -0,0 +1,60 @@
+/*
+ * 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.diff;
+
+import java.math.BigInteger;
+import java.util.List;
+
+import org.junit.Test;
+
+import org.apache.cassandra.diff.DiffJob;
+import org.apache.cassandra.diff.TokenHelper;
+
+import static org.junit.Assert.assertEquals;
+
+public class DiffJobTest
+{
+    @Test
+    public void testSplitsM3P()
+    {
+        splitTestHelper(TokenHelper.forPartitioner("Murmur3Partitioner"));
+    }
+    @Test
+    public void testSplitsRandom()
+    {
+        splitTestHelper(TokenHelper.forPartitioner("RandomPartitioner"));
+    }
+
+    private void splitTestHelper(TokenHelper tokens)
+    {
+        List<DiffJob.Split> splits = DiffJob.calculateSplits(50, 1, tokens);
+        assertEquals(splits.get(0).start, tokens.min());
+        DiffJob.Split prevSplit = null;
+        for (DiffJob.Split split : splits)
+        {
+            if (prevSplit != null)
+                assertEquals(prevSplit.end, split.start.subtract(BigInteger.ONE));
+            prevSplit = split;
+        }
+        assertEquals(splits.get(splits.size() - 1).end, tokens.max());
+        for (int i = 0; i < splits.size(); i++)
+            assertEquals(i, splits.get(i).splitNumber);
+    }
+}
diff --git a/spark-job/src/test/java/org/apache/cassandra/diff/DifferTest.java b/spark-job/src/test/java/org/apache/cassandra/diff/DifferTest.java
new file mode 100644
index 0000000..73677d9
--- /dev/null
+++ b/spark-job/src/test/java/org/apache/cassandra/diff/DifferTest.java
@@ -0,0 +1,111 @@
+/*
+ * 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.diff;
+
+import java.math.BigInteger;
+import java.util.Map;
+import java.util.function.Function;
+
+import com.google.common.base.VerifyException;
+import org.junit.Test;
+
+import com.google.common.collect.Lists;
+
+import org.apache.cassandra.diff.DiffJob;
+import org.apache.cassandra.diff.Differ;
+import org.apache.cassandra.diff.RangeStats;
+import org.apache.cassandra.diff.TokenHelper;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+public class DifferTest {
+
+    @Test(expected = VerifyException.class)
+    public void rejectNullStartOfRange() {
+        Differ.validateRange(null, BigInteger.TEN, TokenHelper.MURMUR3);
+    }
+
+    @Test(expected = VerifyException.class)
+    public void rejectNullEndOfRange() {
+        Differ.validateRange(BigInteger.TEN, null, TokenHelper.MURMUR3);
+    }
+
+    @Test(expected = VerifyException.class)
+    public void rejectWrappingRange() {
+        Differ.validateRange(BigInteger.TEN, BigInteger.ONE, TokenHelper.MURMUR3);
+    }
+
+    @Test(expected = VerifyException.class)
+    public void rejectRangeWithStartLessThanMinMurmurToken() {
+        Differ.validateRange(TokenHelper.MURMUR3.min().subtract(BigInteger.ONE),
+                             BigInteger.TEN,
+                             TokenHelper.MURMUR3);
+    }
+
+    @Test(expected = VerifyException.class)
+    public void rejectRangeWithEndGreaterThanMaxMurmurToken() {
+        Differ.validateRange(BigInteger.ONE,
+                             TokenHelper.MURMUR3.max().add(BigInteger.ONE),
+                             TokenHelper.MURMUR3);
+    }
+
+    @Test
+    public void filterTaskStatusForTables() {
+        // according to the journal:
+        // * t1 is already completed
+        // * t2 is started and has reported some progress, but has not completed
+        // * t3 has not reported any progress
+        DiffJob.Split split = new DiffJob.Split(1, 1, BigInteger.ONE, BigInteger.TEN);
+        Iterable<String> tables = Lists.newArrayList("t1", "t2", "t3");
+        Function<String, DiffJob.TaskStatus> journal = (table) -> {
+            switch (table) {
+                case "t1":
+                    return new DiffJob.TaskStatus(split.end, RangeStats.withValues(6, 6, 6, 6, 6, 6, 6, 6, 6));
+                case "t2":
+                    return new DiffJob.TaskStatus(BigInteger.valueOf(5L), RangeStats.withValues(5, 5, 5, 5, 5, 5, 5, 5, 5));
+                case "t3":
+                    return DiffJob.TaskStatus.EMPTY;
+                default:
+                    throw new AssertionError();
+            }
+        };
+
+        Map<String, DiffJob.TaskStatus> filtered = Differ.filterTables(tables, split, journal, false);
+        assertEquals(2, filtered.keySet().size());
+        assertEquals(RangeStats.withValues(5, 5, 5, 5, 5, 5, 5, 5, 5), filtered.get("t2").stats);
+        assertEquals(BigInteger.valueOf(5L), filtered.get("t2").lastToken);
+        assertEquals(RangeStats.newStats(), filtered.get("t3").stats);
+        assertNull(filtered.get("t3").lastToken);
+
+        // if re-running (part of) a job because of failures or problematic partitions, we want to
+        // ignore the status of completed tasks and re-run them anyway as only specified tokens will
+        // be processed - so t1 should be included now
+        filtered = Differ.filterTables(tables, split, journal, true);
+        assertEquals(3, filtered.keySet().size());
+        assertEquals(RangeStats.withValues(6, 6, 6, 6, 6, 6, 6, 6, 6), filtered.get("t1").stats);
+        assertEquals(split.end, filtered.get("t1").lastToken);
+        assertEquals(RangeStats.withValues(5, 5, 5, 5, 5, 5, 5, 5, 5), filtered.get("t2").stats);
+        assertEquals(BigInteger.valueOf(5L), filtered.get("t2").lastToken);
+        assertEquals(RangeStats.newStats(), filtered.get("t3").stats);
+        assertNull(filtered.get("t3").lastToken);
+    }
+
+}
diff --git a/spark-job/src/test/java/org/apache/cassandra/diff/PartitionComparatorTest.java b/spark-job/src/test/java/org/apache/cassandra/diff/PartitionComparatorTest.java
new file mode 100644
index 0000000..79b3638
--- /dev/null
+++ b/spark-job/src/test/java/org/apache/cassandra/diff/PartitionComparatorTest.java
@@ -0,0 +1,524 @@
+/*
+ * 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.diff;
+
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.net.InetAddress;
+import java.nio.ByteBuffer;
+import java.util.*;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import com.google.common.collect.AbstractIterator;
+import com.google.common.collect.Lists;
+import com.google.common.reflect.TypeToken;
+import org.junit.Test;
+
+import com.datastax.driver.core.*;
+import org.apache.cassandra.diff.PartitionComparator;
+import org.apache.cassandra.diff.PartitionStats;
+import org.apache.cassandra.diff.TableSpec;
+
+import static org.junit.Assert.assertEquals;
+
+public class PartitionComparatorTest {
+
+    @Test
+    public void sourceIsNull() {
+        TableSpec t = spec("table1", names("c1", "c2"), names("v1", "v2"));
+        PartitionComparator comparator = comparator(t, null, rows(row(t, 0, 1, 2, 3)));
+        PartitionStats stats = comparator.call();
+        assertStats(stats, true, true, 0, 0, 0);
+    }
+
+    @Test
+    public void targetIsNull() {
+        TableSpec t = spec("table1", names("c1", "c2"), names("v1", "v2"));
+        PartitionComparator comparator = comparator(t, rows(row(t, 0, 1, 2, 3)), null);
+        PartitionStats stats = comparator.call();
+        assertStats(stats, true, true, 0, 0, 0);
+    }
+
+    @Test
+    public void sourceIsEmpty() {
+        TableSpec t = spec("table1", names("c1", "c2"), names("v1", "v2"));
+        PartitionComparator comparator = comparator(t, rows(), rows(row(t, 0, 1, 2, 3)));
+        PartitionStats stats = comparator.call();
+        assertStats(stats, false, false, 0, 0, 0);
+    }
+
+    @Test
+    public void targetIsEmpty() {
+        TableSpec t = spec("table1", names("c1", "c2"), names("v1", "v2"));
+        PartitionComparator comparator = comparator(t, rows(row(t, 0, 1, 2, 3)), rows());
+        PartitionStats stats = comparator.call();
+        assertStats(stats, false, false, 0, 0, 0);
+    }
+
+    @Test
+    public void sourceAndTargetAreEmpty() {
+        TableSpec t = spec("table1", names("c1", "c2"), names("v1", "v2"));
+        PartitionComparator comparator = comparator(t, rows(), rows());
+        PartitionStats stats = comparator.call();
+        assertStats(stats, false, true, 0, 0, 0);
+    }
+
+    @Test
+    public void sourceContainsExtraRowsAtStart() {
+        TableSpec t = spec("table1", names("c1", "c2"), names("v1", "v2"));
+        PartitionComparator comparator = comparator(t,
+                                                    rows(row(t, 0, 1, 2, 3),
+                                                         row(t, 10, 11, 12, 13)),
+                                                    rows(row(t, 10, 11, 12, 13)));
+        PartitionStats stats = comparator.call();
+        // Comparison fails fast, so bails on the initial mismatch
+        assertStats(stats, false, false, 0, 0, 0);
+    }
+
+    @Test
+    public void targetContainsExtraRowsAtStart() {
+        TableSpec t = spec("table1", names("c1", "c2"), names("v1", "v2"));
+        PartitionComparator comparator = comparator(t,
+                                                    rows(row(t, 10, 11, 12, 13)),
+                                                    rows(row(t, 0, 1, 2, 3),
+                                                         row(t, 10, 11, 12, 13)));
+        PartitionStats stats = comparator.call();
+        // Comparison fails fast, so bails on the initial mismatch
+        assertStats(stats, false, false, 0, 0, 0);
+    }
+
+    @Test
+    public void sourceContainsExtraRowsAtEnd() {
+        TableSpec t = spec("table1", names("c1", "c2"), names("v1", "v2"));
+        PartitionComparator comparator = comparator(t,
+                                                    rows(row(t, 0, 1, 2, 3),
+                                                         row(t, 10, 11, 12, 13)),
+                                                    rows(row(t, 0, 1, 2, 3)));
+        PartitionStats stats = comparator.call();
+        // The fact that the first row & all its v1 & v2 values match should be reflected in the stats
+        assertStats(stats, false, false, 1, 2, 0);
+    }
+
+    @Test
+    public void targetContainsExtraRowsAtEnd() {
+        TableSpec t = spec("table1", names("c1", "c2"), names("v1", "v2"));
+        PartitionComparator comparator = comparator(t,
+                                                    rows(row(t, 0, 1, 2, 3)),
+                                                    rows(row(t, 0, 1, 2, 3),
+                                                         row(t, 10, 11, 12, 13)));
+        PartitionStats stats = comparator.call();
+        // The fact that the first row & all its v1 & v2 values match should be reflected in the stats
+        assertStats(stats, false, false, 1, 2, 0);
+    }
+
+    @Test
+    public void withoutClusteringsAllRowsMatching() {
+        TableSpec t = spec("table1", names(), names("v1", "v2"));
+        PartitionComparator comparator = comparator(t,
+                                                    rows(row(t, 0, 1),
+                                                         row(t, 1, 11),
+                                                         row(t, 2, 12)),
+                                                    rows(row(t, 0, 1),
+                                                         row(t, 1, 11),
+                                                         row(t, 2, 12)));
+        PartitionStats stats = comparator.call();
+        assertStats(stats, false, true, 3, 6, 0);
+    }
+
+    @Test
+    public void singleClusteringAllRowsMatching() {
+        TableSpec t = spec("table1", names("c1"), names("v1", "v2"));
+        PartitionComparator comparator = comparator(t,
+                                                    rows(row(t, 0, 1, 2),
+                                                         row(t, 1, 11, 12),
+                                                         row(t, 2, 21, 22)),
+                                                    rows(row(t, 0, 1, 2),
+                                                         row(t, 1, 11, 12),
+                                                         row(t, 2, 21, 22)));
+        PartitionStats stats = comparator.call();
+        assertStats(stats, false, true, 3, 6, 0);
+    }
+
+    @Test
+    public void multipleClusteringAllRowsMatching() {
+        TableSpec t = spec("table1", names("c1", "c2"), names("v1", "v2"));
+        PartitionComparator comparator = comparator(t,
+                                                    rows(row(t, 0, 1, 2, 3),
+                                                         row(t, 1, 11, 12, 13),
+                                                         row(t, 2, 21, 22, 23)),
+                                                    rows(row(t, 0, 1, 2, 3),
+                                                         row(t, 1, 11, 12, 13),
+                                                         row(t, 2, 21, 22, 23)));
+        PartitionStats stats = comparator.call();
+        assertStats(stats, false, true, 3, 6, 0);
+    }
+
+    @Test
+    public void withoutClusteringsWithMismatches() {
+        TableSpec t = spec("table1", names(), names("v1", "v2"));
+        PartitionComparator comparator = comparator(t,
+                                                    rows(row(t, 0, 1),
+                                                         row(t, 1, 11),
+                                                         row(t, 2, 12)),
+                                                    rows(row(t, 0, 1),
+                                                         row(t, 1, 1100),
+                                                         row(t, 2, 1200)));
+        PartitionStats stats = comparator.call();
+        assertStats(stats, false, true, 3, 4, 2);
+    }
+
+    @Test
+    public void singleClusteringWithMismatches() {
+        TableSpec t = spec("table1", names("c1"), names("v1", "v2"));
+        PartitionComparator comparator = comparator(t,
+                                                    rows(row(t, 0, 1, 2),
+                                                         row(t, 1, 11, 21),
+                                                         row(t, 2, 12, 22)),
+                                                    rows(row(t, 0, 1, 20),
+                                                         row(t, 1, 1100, 21),
+                                                         row(t, 2, 12, 1200)));
+        PartitionStats stats = comparator.call();
+        assertStats(stats, false, true, 3, 3, 3);
+    }
+
+    private void assertStats(PartitionStats stats,
+                             boolean skipped,
+                             boolean clusteringsMatch,
+                             int matchedRows,
+                             int matchedValues,
+                             int mismatchedValues) {
+        assertEquals(skipped, stats.skipped);
+        assertEquals(clusteringsMatch, stats.allClusteringsMatch);
+        assertEquals(matchedRows, stats.matchedRows);
+        assertEquals(matchedValues, stats.matchedValues);
+        assertEquals(mismatchedValues, stats.mismatchedValues);
+    }
+
+    PartitionComparator comparator(TableSpec table, Iterator<Row> source, Iterator<Row> target) {
+        return new PartitionComparator(table, source, target);
+    }
+
+    List<String> names(String...names) {
+        return Lists.newArrayList(names);
+    }
+
+    TableSpec spec(String table, List<String> clusteringColumns, List<String> regularColumns) {
+        return new TableSpec(table, columns(clusteringColumns), columns(regularColumns));
+    }
+
+    List<ColumnMetadata> columns(List<String> names) {
+        return names.stream().map(ColumnMetadataHelper::column).collect(Collectors.toList());
+    }
+
+    Iterator<Row> rows(Row...rows) {
+        return new AbstractIterator<Row>() {
+            int i = 0;
+            protected Row computeNext() {
+                if (i < rows.length)
+                    return rows[i++];
+                return endOfData();
+            }
+        };
+    }
+
+    Row row(TableSpec table, Object...values) {
+        return new TestRow(Stream.concat(table.getClusteringColumns().stream(),
+                                         table.getRegularColumns().stream())
+                                 .map(ColumnMetadata::getName).toArray(String[]::new),
+                           values);
+    }
+
+    static class TestRow implements Row {
+        private final String[] names;
+        private final Object[] values;
+
+        TestRow(String[] names, Object[] values) {
+            if (names.length != values.length)
+                throw new IllegalArgumentException(String.format("Number of column names (%d) doesn't " +
+                                                                 "match number of values(%d)",
+                                                                 names.length, values.length));
+            this.names = names;
+            this.values = values;
+        }
+
+        // Only getObject(String) is used by PartitionComparator
+        public Object getObject(String s) {
+            for (int i=0; i < names.length; i++)
+                if (names[i].equals(s))
+                    return values[i];
+
+            throw new IllegalArgumentException(s + " is not a column defined in this metadata");
+        }
+
+        public boolean isNull(String s) {
+            throw new UnsupportedOperationException();
+        }
+
+        public boolean getBool(String s) {
+            throw new UnsupportedOperationException();
+        }
+
+        public byte getByte(String s) {
+            throw new UnsupportedOperationException();
+        }
+
+        public short getShort(String s) {
+            throw new UnsupportedOperationException();
+        }
+
+        public int getInt(String s) {
+            throw new UnsupportedOperationException();
+        }
+
+        public long getLong(String s) {
+            throw new UnsupportedOperationException();
+        }
+
+        public Date getTimestamp(String s) {
+            throw new UnsupportedOperationException();
+        }
+
+        public LocalDate getDate(String s) {
+            throw new UnsupportedOperationException();
+        }
+
+        public long getTime(String s) {
+            throw new UnsupportedOperationException();
+        }
+
+        public float getFloat(String s) {
+            throw new UnsupportedOperationException();
+        }
+
+        public double getDouble(String s) {
+            throw new UnsupportedOperationException();
+        }
+
+        public ByteBuffer getBytesUnsafe(String s) {
+            throw new UnsupportedOperationException();
+        }
+
+        public ByteBuffer getBytes(String s) {
+            throw new UnsupportedOperationException();
+        }
+
+        public String getString(String s) {
+            throw new UnsupportedOperationException();
+        }
+
+        public BigInteger getVarint(String s) {
+            throw new UnsupportedOperationException();
+        }
+
+        public BigDecimal getDecimal(String s) {
+            throw new UnsupportedOperationException();
+        }
+
+        public UUID getUUID(String s) {
+            throw new UnsupportedOperationException();
+        }
+
+        public InetAddress getInet(String s) {
+            throw new UnsupportedOperationException();
+        }
+
+        public <T> List<T> getList(String s, Class<T> aClass) {
+            throw new UnsupportedOperationException();
+        }
+
+        public <T> List<T> getList(String s, TypeToken<T> typeToken) {
+            throw new UnsupportedOperationException();
+        }
+
+        public <T> Set<T> getSet(String s, Class<T> aClass) {
+            throw new UnsupportedOperationException();
+        }
+
+        public <T> Set<T> getSet(String s, TypeToken<T> typeToken) {
+            throw new UnsupportedOperationException();
+        }
+
+        public <K, V> Map<K, V> getMap(String s, Class<K> aClass, Class<V> aClass1) {
+            throw new UnsupportedOperationException();
+        }
+
+        public <K, V> Map<K, V> getMap(String s, TypeToken<K> typeToken, TypeToken<V> typeToken1) {
+            throw new UnsupportedOperationException();
+        }
+
+        public UDTValue getUDTValue(String s) {
+            throw new UnsupportedOperationException();
+        }
+
+        public TupleValue getTupleValue(String s) {
+            throw new UnsupportedOperationException();
+        }
+
+        public <T> T get(String s, Class<T> aClass) {
+            throw new UnsupportedOperationException();
+        }
+
+        public <T> T get(String s, TypeToken<T> typeToken) {
+            throw new UnsupportedOperationException();
+        }
+
+        public <T> T get(String s, TypeCodec<T> typeCodec) {
+            throw new UnsupportedOperationException();
+        }
+
+        public ColumnDefinitions getColumnDefinitions() {
+            throw new UnsupportedOperationException();
+        }
+
+        public Token getToken(int i) {
+            throw new UnsupportedOperationException();
+        }
+
+        public Token getToken(String s) {
+            throw new UnsupportedOperationException();
+        }
+
+        public Token getPartitionKeyToken() {
+            throw new UnsupportedOperationException();
+        }
+
+        public boolean isNull(int i) {
+            throw new UnsupportedOperationException();
+        }
+
+        public boolean getBool(int i) {
+            throw new UnsupportedOperationException();
+        }
+
+        public byte getByte(int i) {
+            throw new UnsupportedOperationException();
+        }
+
+        public short getShort(int i) {
+            throw new UnsupportedOperationException();
+        }
+
+        public int getInt(int i) {
+            throw new UnsupportedOperationException();
+        }
+
+        public long getLong(int i) {
+            throw new UnsupportedOperationException();
+        }
+
+        public Date getTimestamp(int i) {
+            throw new UnsupportedOperationException();
+        }
+
+        public LocalDate getDate(int i) {
+            throw new UnsupportedOperationException();
+        }
+
+        public long getTime(int i) {
+            throw new UnsupportedOperationException();
+        }
+
+        public float getFloat(int i) {
+            throw new UnsupportedOperationException();
+        }
+
+        public double getDouble(int i) {
+            throw new UnsupportedOperationException();
+        }
+
+        public ByteBuffer getBytesUnsafe(int i) {
+            throw new UnsupportedOperationException();
+        }
+
+        public ByteBuffer getBytes(int i) {
+            throw new UnsupportedOperationException();
+        }
+
+        public String getString(int i) {
+            throw new UnsupportedOperationException();
+        }
+
+        public BigInteger getVarint(int i) {
+            throw new UnsupportedOperationException();
+        }
+
+        public BigDecimal getDecimal(int i) {
+            throw new UnsupportedOperationException();
+        }
+
+        public UUID getUUID(int i) {
+            throw new UnsupportedOperationException();
+        }
+
+        public InetAddress getInet(int i) {
+            throw new UnsupportedOperationException();
+        }
+
+        public <T> List<T> getList(int i, Class<T> aClass) {
+            throw new UnsupportedOperationException();
+        }
+
+        public <T> List<T> getList(int i, TypeToken<T> typeToken) {
+            throw new UnsupportedOperationException();
+        }
+
+        public <T> Set<T> getSet(int i, Class<T> aClass) {
+            throw new UnsupportedOperationException();
+        }
+
+        public <T> Set<T> getSet(int i, TypeToken<T> typeToken) {
+            throw new UnsupportedOperationException();
+        }
+
+        public <K, V> Map<K, V> getMap(int i, Class<K> aClass, Class<V> aClass1) {
+            throw new UnsupportedOperationException();
+        }
+
+        public <K, V> Map<K, V> getMap(int i, TypeToken<K> typeToken, TypeToken<V> typeToken1) {
+            throw new UnsupportedOperationException();
+        }
+
+        public UDTValue getUDTValue(int i) {
+            throw new UnsupportedOperationException();
+        }
+
+        public TupleValue getTupleValue(int i) {
+            throw new UnsupportedOperationException();
+        }
+
+        public Object getObject(int i) {
+            throw new UnsupportedOperationException();
+        }
+
+        public <T> T get(int i, Class<T> aClass) {
+            throw new UnsupportedOperationException();
+        }
+
+        public <T> T get(int i, TypeToken<T> typeToken) {
+            throw new UnsupportedOperationException();
+        }
+
+        public <T> T get(int i, TypeCodec<T> typeCodec) {
+            throw new UnsupportedOperationException();
+        }
+    }
+
+}
diff --git a/spark-job/src/test/java/org/apache/cassandra/diff/RangeComparatorTest.java b/spark-job/src/test/java/org/apache/cassandra/diff/RangeComparatorTest.java
new file mode 100644
index 0000000..0484212
--- /dev/null
+++ b/spark-job/src/test/java/org/apache/cassandra/diff/RangeComparatorTest.java
@@ -0,0 +1,642 @@
+/*
+ * 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.diff;
+
+import java.math.BigInteger;
+import java.nio.ByteBuffer;
+import java.util.*;
+import java.util.concurrent.CountDownLatch;
+import java.util.function.BiConsumer;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.LongStream;
+
+import com.google.common.collect.*;
+import org.junit.Test;
+
+import com.codahale.metrics.MetricRegistry;
+import com.datastax.driver.core.DataType;
+import com.datastax.driver.core.ProtocolVersion;
+import com.datastax.driver.core.Token;
+import org.jetbrains.annotations.NotNull;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import static org.apache.cassandra.diff.TestUtils.assertThreadWaits;
+
+public class RangeComparatorTest {
+
+    private Multimap<BigInteger, Throwable> errors = HashMultimap.create();
+    private BiConsumer<Throwable, BigInteger> errorReporter = (e, t) -> errors.put(t, e);
+    private Multimap<BigInteger, MismatchType> mismatches = HashMultimap.create();
+    private BiConsumer<MismatchType, BigInteger> mismatchReporter = (m, t) -> mismatches.put(t, m);
+    private Multimap<BigInteger, RangeStats> journal= HashMultimap.create();
+    private BiConsumer<RangeStats, BigInteger> progressReporter = (r, t) -> journal.put(t, copyOf(r));
+    private Set<BigInteger> comparedPartitions = new HashSet<>();
+    private ComparisonExecutor executor = ComparisonExecutor.newExecutor(1, new MetricRegistry());
+
+    @Test
+    public void emptyRange() {
+        RangeComparator comparator = comparator(context(100L, 100L));
+        RangeStats stats = comparator.compare(keys(), keys(), this::alwaysMatch);
+        assertTrue(stats.isEmpty());
+        assertNothingReported(errors, mismatches, journal);
+        assertCompared();
+    }
+
+    @Test
+    public void sourceAndTargetKeysAreEmpty() {
+        RangeComparator comparator = comparator(context(0L, 100L));
+        RangeStats stats = comparator.compare(keys(), keys(), this::alwaysMatch);
+        assertTrue(stats.isEmpty());
+        assertNothingReported(errors, mismatches, journal);
+        assertCompared();
+    }
+
+    @Test
+    public void partitionPresentOnlyInSource() {
+        RangeComparator comparator = comparator(context(0L, 100L));
+        RangeStats stats = comparator.compare(keys(0, 1, 2), keys(0, 1), this::alwaysMatch);
+        assertFalse(stats.isEmpty());
+        assertEquals(1, stats.getOnlyInSource());
+        assertEquals(0, stats.getOnlyInTarget());
+        assertEquals(2, stats.getMatchedPartitions());
+        assertReported(2, MismatchType.ONLY_IN_SOURCE, mismatches);
+        assertNothingReported(errors, journal);
+        assertCompared(0, 1);
+    }
+
+    @Test
+    public void partitionPresentOnlyInTarget() {
+        RangeComparator comparator = comparator(context(0L, 100L));
+        RangeStats stats = comparator.compare(keys(7, 8), keys(7, 8, 9), this::alwaysMatch);
+        assertFalse(stats.isEmpty());
+        assertEquals(0, stats.getOnlyInSource());
+        assertEquals(1, stats.getOnlyInTarget());
+        assertEquals(2, stats.getMatchedPartitions());
+        assertReported(9, MismatchType.ONLY_IN_TARGET, mismatches);
+        assertNothingReported(errors, journal);
+        assertCompared(7, 8);
+    }
+
+    @Test
+    public void multipleOnlyInSource() {
+        RangeComparator comparator = comparator(context(0L, 100L));
+        RangeStats stats = comparator.compare(keys(4, 5, 6, 7, 8), keys(4, 5), this::alwaysMatch);
+        assertFalse(stats.isEmpty());
+        assertEquals(3, stats.getOnlyInSource());
+        assertEquals(0, stats.getOnlyInTarget());
+        assertEquals(2, stats.getMatchedPartitions());
+        assertReported(6, MismatchType.ONLY_IN_SOURCE, mismatches);
+        assertReported(7, MismatchType.ONLY_IN_SOURCE, mismatches);
+        assertReported(8, MismatchType.ONLY_IN_SOURCE, mismatches);
+        assertNothingReported(errors, journal);
+        assertCompared(4, 5);
+    }
+
+    @Test
+    public void multipleOnlyInTarget() {
+        RangeComparator comparator = comparator(context(0L, 100L));
+        RangeStats stats = comparator.compare(keys(4, 5), keys(4, 5, 6, 7, 8), this::alwaysMatch);
+        assertFalse(stats.isEmpty());
+        assertEquals(0, stats.getOnlyInSource());
+        assertEquals(3, stats.getOnlyInTarget());
+        assertEquals(2, stats.getMatchedPartitions());
+        assertReported(6, MismatchType.ONLY_IN_TARGET, mismatches);
+        assertReported(7, MismatchType.ONLY_IN_TARGET, mismatches);
+        assertReported(8, MismatchType.ONLY_IN_TARGET, mismatches);
+        assertNothingReported(errors, journal);
+        assertCompared(4, 5);
+    }
+
+    @Test
+    public void multipleOnlyInBoth() {
+        RangeComparator comparator = comparator(context(0L, 100L));
+        RangeStats stats = comparator.compare(keys(0, 1, 3, 5, 7, 9), keys(0, 2, 4, 6, 8, 9), this::alwaysMatch);
+        assertFalse(stats.isEmpty());
+        assertEquals(4, stats.getOnlyInSource());
+        assertEquals(4, stats.getOnlyInTarget());
+        assertEquals(2, stats.getMatchedPartitions());
+        assertReported(1, MismatchType.ONLY_IN_SOURCE, mismatches);
+        assertReported(3, MismatchType.ONLY_IN_SOURCE, mismatches);
+        assertReported(5, MismatchType.ONLY_IN_SOURCE, mismatches);
+        assertReported(7, MismatchType.ONLY_IN_SOURCE, mismatches);
+        assertReported(2, MismatchType.ONLY_IN_TARGET, mismatches);
+        assertReported(4, MismatchType.ONLY_IN_TARGET, mismatches);
+        assertReported(6, MismatchType.ONLY_IN_TARGET, mismatches);
+        assertReported(8, MismatchType.ONLY_IN_TARGET, mismatches);
+        assertNothingReported(errors, journal);
+        assertCompared(0, 9);
+    }
+
+    @Test
+    public void sourceKeysIsEmpty() {
+        RangeComparator comparator = comparator(context(0L, 100L));
+        RangeStats stats = comparator.compare(keys(), keys(4, 5), this::alwaysMatch);
+        assertFalse(stats.isEmpty());
+        assertEquals(0, stats.getOnlyInSource());
+        assertEquals(2, stats.getOnlyInTarget());
+        assertEquals(0, stats.getMatchedPartitions());
+        assertReported(4, MismatchType.ONLY_IN_TARGET, mismatches);
+        assertReported(5, MismatchType.ONLY_IN_TARGET, mismatches);
+        assertNothingReported(errors, journal);
+        assertCompared();
+    }
+
+    @Test
+    public void targetKeysIsEmpty() {
+        RangeComparator comparator = comparator(context(0L, 100L));
+        RangeStats stats = comparator.compare(keys(4, 5), keys(), this::alwaysMatch);
+        assertFalse(stats.isEmpty());
+        assertEquals(2, stats.getOnlyInSource());
+        assertEquals(0, stats.getOnlyInTarget());
+        assertEquals(0, stats.getMatchedPartitions());
+        assertReported(4, MismatchType.ONLY_IN_SOURCE, mismatches);
+        assertReported(5, MismatchType.ONLY_IN_SOURCE, mismatches);
+        assertNothingReported(errors, journal);
+        assertCompared();
+    }
+
+    @Test
+    public void skipComparisonOfDisallowedTokens() {
+        RangeComparator comparator = comparator(context(0L, 100L, 2, 3, 4));
+        RangeStats stats = comparator.compare(keys(0, 1, 2, 3, 4, 5, 6),
+                                              keys(0, 1, 2, 3, 4, 5, 6),
+                                              this::alwaysMatch);
+        assertFalse(stats.isEmpty());
+        assertEquals(0, stats.getOnlyInSource());
+        assertEquals(0, stats.getOnlyInTarget());
+        assertEquals(4, stats.getMatchedPartitions());
+        assertEquals(3, stats.getSkippedPartitions());
+        assertNothingReported(errors,  mismatches, journal);
+        assertCompared(0, 1, 5 , 6);
+    }
+
+    @Test
+    public void handleErrorReadingFirstSourceKey() {
+        RuntimeException toThrow = new RuntimeException("Test");
+        testErrorReadingKey(keys(0, toThrow, 0, 1, 2), keys(0, 1, 2), toThrow);
+        assertCompared();
+    }
+
+    @Test
+    public void handleErrorReadingFirstTargetKey() {
+        RuntimeException toThrow = new RuntimeException("Test");
+        testErrorReadingKey(keys(0, 1, 2), keys(0, toThrow, 0, 1, 2), toThrow);
+        assertCompared();
+    }
+
+    @Test
+    public void handleErrorReadingSourceKey() {
+        RuntimeException toThrow = new RuntimeException("Test");
+        testErrorReadingKey(keys(1, toThrow, 0, 1, 2), keys(0, 1, 2), toThrow);
+        assertCompared(0);
+    }
+
+    @Test
+    public void handleErrorReadingTargetKey() {
+        RuntimeException toThrow = new RuntimeException("Test");
+        testErrorReadingKey(keys(0, 1, 2), keys(1, toThrow, 0, 1, 2), toThrow);
+        assertCompared(0);
+    }
+
+    @Test
+    public void handleReadingLastSourceKey() {
+        RuntimeException toThrow = new RuntimeException("Test");
+        testErrorReadingKey(keys(2, toThrow, 0, 1, 2), keys(0, 1, 2), toThrow);
+        assertCompared(0, 1);
+    }
+
+    @Test
+    public void handleReadingLastTargetKey() {
+        RuntimeException toThrow = new RuntimeException("Test");
+        testErrorReadingKey(keys(0, 1, 2), keys(2, toThrow, 0, 1, 2), toThrow);
+        assertCompared(0, 1);
+    }
+
+    @Test
+    public void handleErrorConstructingFirstTask() {
+        RuntimeException expected = new RuntimeException("Test");
+        RangeStats stats = testTaskError(throwDuringConstruction(0, expected));
+        assertEquals(1, stats.getErrorPartitions());
+        assertCompared(1, 2);
+        assertReported(0, expected, errors);
+    }
+
+    @Test
+    public void handleErrorConstructingTask() {
+        RuntimeException expected = new RuntimeException("Test");
+        RangeStats stats = testTaskError(throwDuringConstruction(1, expected));
+        assertEquals(1, stats.getErrorPartitions());
+        assertCompared(0, 2);
+        assertReported(1, expected, errors);
+    }
+
+    @Test
+    public void handleErrorConstructingLastTask() {
+        RuntimeException expected = new RuntimeException("Test");
+        RangeStats stats = testTaskError(throwDuringConstruction(2, expected));
+        assertEquals(1, stats.getErrorPartitions());
+        assertCompared(0, 1);
+        assertReported(2, expected, errors);
+    }
+
+    @Test
+    public void handleTaskErrorOnFirstPartition() {
+        RuntimeException expected = new RuntimeException("Test");
+        RangeStats stats = testTaskError(throwDuringExecution(expected, 0));
+        assertEquals(1, stats.getErrorPartitions());
+        assertCompared(1, 2);
+        assertReported(0, expected, errors);
+    }
+
+    @Test
+    public void handleTaskErrorOnPartition() {
+        RuntimeException expected = new RuntimeException("Test");
+        RangeStats stats = testTaskError(throwDuringExecution(expected, 1));
+        assertEquals(1, stats.getErrorPartitions());
+        assertCompared(0, 2);
+        assertReported(1, expected, errors);
+    }
+
+    @Test
+    public void handleTaskErrorOnLastPartition() {
+        RuntimeException expected = new RuntimeException("Test");
+        RangeStats stats = testTaskError(throwDuringExecution(expected, 2));
+        assertEquals(1, stats.getErrorPartitions());
+        assertCompared(0, 1);
+        assertReported(2, expected, errors);
+    }
+
+    @Test
+    public void checkpointEveryTenPartitions() {
+        RangeComparator comparator = comparator(context(0L, 100L));
+        comparator.compare(keys(LongStream.range(0, 25).toArray()),
+                           keys(LongStream.range(0, 25).toArray()),
+                           this::alwaysMatch);
+        assertReported(9, RangeStats.withValues(10, 0L, 0L, 0L, 0L, 0L, 0L, 0L, 0L), journal);
+        assertReported(19, RangeStats.withValues(20, 0L, 0L, 0L, 0L, 0L, 0L, 0L, 0L), journal);
+        assertEquals(2, journal.keySet().size());
+    }
+
+    @Test
+    public void recordHighestSeenPartitionWhenTasksCompleteOutOfOrder() {
+        // every 10 partitions, the highest seen token is reported to the journal. Here,
+        // randomise the iteration of the keys to simulate tasks completing out of order
+        RangeComparator comparator = comparator(context(0L, 100L));
+        long[] tokens = new long[] {2, 8, 1, 4, 100, 3, 5, 7, 6, 9};
+        comparator.compare(keys(tokens),
+                           keys(tokens),
+                           this::alwaysMatch);
+        assertReported(100, RangeStats.withValues(10, 0L, 0L, 0L, 0L, 0L, 0L, 0L, 0L), journal);
+        assertEquals(1, journal.keySet().size());
+        assertNothingReported(errors, mismatches);
+        assertCompared(tokens);
+    }
+
+    @Test
+    public void recordHighestSeenPartitionWhenTasksCompleteOutOfOrderWithErrors() {
+        // 2 partitions will error during comparison, but after 10 successful comparisons
+        // we'll still report progress to the journal
+        RuntimeException toThrow = new RuntimeException("Test");
+        RangeComparator comparator = comparator(context(0L, 100L));
+        long[] tokens = new long[] {2, 8, 1, 11, 4, 100, 3, 5, 7, 6, 9, 0};
+        comparator.compare(keys(tokens), keys(tokens), throwDuringExecution(toThrow, 3, 2));
+        assertReported(100, RangeStats.withValues(10, 0L, 2L, 0L, 0L, 0L, 0L, 0L, 0L), journal);
+        assertEquals(1, journal.keySet().size());
+        assertReported(2, toThrow, errors);
+        assertReported(3, toThrow, errors);
+        assertEquals(2, errors.keySet().size());
+        assertNothingReported(mismatches);
+        // the erroring tasks don't get counted in the test as compared
+        assertCompared(8, 1, 11, 4, 100, 5, 7, 6, 9, 0);
+    }
+
+    @Test
+    public void rowLevelMismatchIncrementsPartitionMismatches() {
+        RangeComparator comparator = comparator(context(0L, 100L));
+        RangeStats stats = comparator.compare(keys(0, 1, 2), keys(0, 1 ,2), this::rowMismatch);
+        assertEquals(0, stats.getMatchedPartitions());
+        assertEquals(3, stats.getMismatchedPartitions());
+        assertEquals(0, stats.getMatchedValues());
+        assertEquals(0, stats.getMismatchedValues());
+        assertNothingReported(errors, journal);
+        assertReported(0, MismatchType.PARTITION_MISMATCH, mismatches);
+        assertReported(1, MismatchType.PARTITION_MISMATCH, mismatches);
+        assertReported(2, MismatchType.PARTITION_MISMATCH, mismatches);
+        assertCompared(0, 1, 2);
+    }
+
+    @Test
+    public void valueMismatchIncrementsPartitionMismatches() {
+        RangeComparator comparator = comparator(context(0L, 100L));
+        RangeStats stats = comparator.compare(keys(0, 1, 2), keys(0, 1 ,2), this::valuesMismatch);
+        assertEquals(0, stats.getMatchedPartitions());
+        assertEquals(3, stats.getMismatchedPartitions());
+        assertEquals(0, stats.getMatchedValues());
+        assertEquals(30, stats.getMismatchedValues());
+        assertNothingReported(errors, journal);
+        assertReported(0, MismatchType.PARTITION_MISMATCH, mismatches);
+        assertReported(1, MismatchType.PARTITION_MISMATCH, mismatches);
+        assertReported(2, MismatchType.PARTITION_MISMATCH, mismatches);
+        assertCompared(0, 1, 2);
+    }
+
+    @Test
+    public void matchingPartitionIncrementsCount() {
+        RangeComparator comparator = comparator(context(0L, 100L));
+        RangeStats stats = comparator.compare(keys(0, 1, 2), keys(0, 1 ,2), this::alwaysMatch);
+        assertEquals(3, stats.getMatchedPartitions());
+        assertEquals(0, stats.getMismatchedPartitions());
+        assertEquals(0, stats.getMatchedValues());
+        assertEquals(0, stats.getMismatchedValues());
+        assertNothingReported(errors, mismatches, journal);
+        assertCompared(0, 1, 2);
+    }
+
+    @Test
+    public void waitForAllInFlightTasksToComplete() throws InterruptedException {
+        CountDownLatch taskSubmissions = new CountDownLatch(2);
+        List<CountDownLatch> taskGates = Lists.newArrayList(new CountDownLatch(1), new CountDownLatch(1));
+        RangeComparator comparator = comparator(context(0L, 100L));
+
+        Thread t = new Thread(() -> comparator.compare(keys(0, 1),
+                                                       keys(0, 1),
+                                                       waitUntilNotified(taskSubmissions, taskGates)),
+                              "CallingThread");
+        t.setDaemon(true);
+        t.start();
+
+        // wait for both tasks to be submitted then check that the calling thread enters a waiting state
+        taskSubmissions.await();
+        assertThreadWaits(t);
+
+        // let the first task complete and check that the calling thread is still waiting
+        taskGates.get(0).countDown();
+        assertThreadWaits(t);
+
+        // let the second task run and wait for the caller to terminate
+        taskGates.get(1).countDown();
+        t.join();
+
+        assertNothingReported(errors, mismatches, journal);
+        assertCompared(0, 1);
+    }
+
+    private RangeStats testTaskError(Function<PartitionKey, PartitionComparator> taskSupplier) {
+        RangeComparator comparator = comparator(context(0L, 100L));
+        Iterator<PartitionKey> sourceKeys = keys(0, 1, 2);
+        Iterator<PartitionKey> targetKeys = keys(0, 1, 2);
+        RangeStats stats = comparator.compare(sourceKeys, targetKeys, taskSupplier);
+        assertNothingReported(mismatches, journal);
+        return stats;
+    }
+
+    private void testErrorReadingKey(Iterator<PartitionKey> sourceKeys,
+                                     Iterator<PartitionKey> targetKeys,
+                                     Exception expected) {
+
+        RangeComparator comparator = comparator(context(0L, 100L));
+        try {
+            comparator.compare(sourceKeys, targetKeys, this::alwaysMatch);
+            fail("Expected exception " + expected.getLocalizedMessage());
+        } catch (Exception e) {
+            assertEquals(expected, e.getCause());
+        }
+        assertNothingReported(errors,  mismatches, journal);
+    }
+
+    private void assertCompared(long...tokens) {
+        assertEquals(comparedPartitions.size(), tokens.length);
+        for(long t : tokens)
+            assertTrue(comparedPartitions.contains(BigInteger.valueOf(t)));
+    }
+
+    private void assertNothingReported(Multimap...reported) {
+        for (Multimap m : reported)
+            assertTrue(m.isEmpty());
+    }
+
+    private <T> void assertReported(long token, T expected, Multimap<BigInteger, T> reported) {
+        Collection<T> values = reported.get(BigInteger.valueOf(token));
+        assertEquals(1, values.size());
+        assertEquals(expected, values.iterator().next());
+    }
+
+    Iterator<PartitionKey> keys(long throwAtToken, RuntimeException e, long...tokens) {
+        return new AbstractIterator<PartitionKey>() {
+            int i = 0;
+            protected PartitionKey computeNext() {
+                if (i < tokens.length) {
+                    long t = tokens[i++];
+                    if (t == throwAtToken)
+                        throw e;
+
+                    return key(t);
+                }
+                return endOfData();
+            }
+        };
+    }
+
+    Iterator<PartitionKey> keys(long...tokens) {
+        return new AbstractIterator<PartitionKey>() {
+            int i = 0;
+            protected PartitionKey computeNext() {
+                if (i < tokens.length)
+                    return key(tokens[i++]);
+                return endOfData();
+            }
+        };
+    }
+
+    // yield a PartitionComparator which always concludes that partitions being compared are identical
+    PartitionComparator alwaysMatch(PartitionKey key) {
+        return new PartitionComparator(null, null, null) {
+            public PartitionStats call() {
+                comparedPartitions.add(key.getTokenAsBigInteger());
+                return new PartitionStats();
+            }
+        };
+    }
+
+    // yield a PartitionComparator which always determines that the partitions have a row-level mismatch
+    PartitionComparator rowMismatch(PartitionKey key) {
+        return new PartitionComparator(null, null,  null) {
+            public PartitionStats call() {
+                comparedPartitions.add(key.getTokenAsBigInteger());
+                PartitionStats stats = new PartitionStats();
+                stats.allClusteringsMatch = false;
+                return stats;
+            }
+        };
+    }
+
+    // yield a PartitionComparator which always determines that the partitions have a 10 mismatching values
+    PartitionComparator valuesMismatch(PartitionKey key) {
+        return new PartitionComparator(null, null,  null) {
+            public PartitionStats call() {
+                comparedPartitions.add(key.getTokenAsBigInteger());
+                PartitionStats stats = new PartitionStats();
+                stats.mismatchedValues = 10;
+                return stats;
+            }
+        };
+    }
+
+    // yield a function which throws when creating a PartitionComparator for a give token
+    // simulates an error reading the source or target partition from the source or target
+    // cluster when constructing the task
+    Function<PartitionKey, PartitionComparator> throwDuringConstruction(long throwAt, RuntimeException toThrow) {
+        return (key) -> {
+            BigInteger t = key.getTokenAsBigInteger();
+            if (t.longValue() == throwAt)
+                throw toThrow;
+
+            return alwaysMatch(key);
+        };
+    }
+
+    // yields a PartitionComparator which throws the supplied exception when the token matches
+    // the one specified to simulate an error when processing the comparison
+    Function<PartitionKey, PartitionComparator> throwDuringExecution(RuntimeException toThrow, long...throwAt) {
+        return (key) -> {
+            BigInteger t = key.getTokenAsBigInteger();
+            return new PartitionComparator(null, null, null) {
+                public PartitionStats call() {
+                    for (long shouldThrow : throwAt)
+                        if (t.longValue() == shouldThrow)
+                           throw toThrow;
+
+                    comparedPartitions.add(t);
+                    return new PartitionStats();
+                }
+            };
+        };
+    }
+
+    // yields a PartitionComparator which waits on a CountDownLatch before returning from call(). The latches
+    // are supplied in an iterator and each successive task yielded uses the next supplied latch so that callers
+    // can control the rate of task progress
+    // The other supplied latch, firstTaskStarted, is used to signal to the caller that execution of the first
+    // task has started, so the test doesn't complete before this happens
+    Function<PartitionKey, PartitionComparator> waitUntilNotified(final CountDownLatch taskSubmissions,
+                                                                  final List<CountDownLatch> taskGates) {
+        final Iterator<CountDownLatch> gateIter = taskGates.iterator();
+        return (key) -> {
+
+            BigInteger t = key.getTokenAsBigInteger();
+            taskSubmissions.countDown();
+
+            return new PartitionComparator(null, null, null) {
+                public PartitionStats call() {
+                    if (!gateIter.hasNext())
+                        fail("Expected a latch to control task progress");
+
+                    try {
+                        gateIter.next().await();
+                    }
+                    catch (InterruptedException e) {
+                        e.printStackTrace();
+                        fail("Interrupted");
+                    }
+                    comparedPartitions.add(t);
+                    return new PartitionStats();
+                }
+            };
+        };
+    }
+
+    PartitionKey key(long token) {
+        return new TestPartitionKey(token);
+    }
+
+    RangeComparator comparator(DiffContext context) {
+        return new RangeComparator(context, errorReporter, mismatchReporter, progressReporter, executor);
+    }
+
+    DiffContext context(long startToken, long endToken, long...disallowedTokens) {
+        return new TestContext(BigInteger.valueOf(startToken),
+                               BigInteger.valueOf(endToken),
+                               new SpecificTokens(Arrays.stream(disallowedTokens)
+                                                        .mapToObj(BigInteger::valueOf)
+                                                        .collect(Collectors.toSet()),
+                                                  SpecificTokens.Modifier.REJECT));
+    }
+
+    DiffContext context(long startToken, long endToken) {
+        return new TestContext(BigInteger.valueOf(startToken), BigInteger.valueOf(endToken), SpecificTokens.NONE);
+    }
+
+    RangeStats copyOf(RangeStats stats) {
+        return RangeStats.withValues(stats.getMatchedPartitions(),
+                                     stats.getMismatchedPartitions(),
+                                     stats.getErrorPartitions(),
+                                     stats.getSkippedPartitions(),
+                                     stats.getOnlyInSource(),
+                                     stats.getOnlyInTarget(),
+                                     stats.getMatchedRows(),
+                                     stats.getMatchedValues(),
+                                     stats.getMismatchedValues());
+    }
+
+    static class TestPartitionKey extends PartitionKey {
+        final Token token;
+
+        TestPartitionKey(final long tokenValue) {
+            super(null);
+            token = new Token() {
+
+                public DataType getType() {
+                    return DataType.bigint();
+                }
+
+                public Object getValue() {
+                    return tokenValue;
+                }
+
+                public ByteBuffer serialize(ProtocolVersion protocolVersion) {
+                    return null;
+                }
+
+                public int compareTo(@NotNull Token o) {
+                    assert o.getValue() instanceof Long;
+                    return Long.compare(tokenValue, (long)o.getValue());
+                }
+            };
+        }
+
+        protected Token getToken() {
+            return token;
+        }
+    }
+
+    static class TestContext extends DiffContext {
+
+        public TestContext(BigInteger startToken,
+                           BigInteger endToken,
+                           SpecificTokens specificTokens) {
+            super(null, null, null, null, startToken, endToken, specificTokens, 0.0);
+        }
+    }
+}
diff --git a/spark-job/src/test/java/org/apache/cassandra/diff/TestUtils.java b/spark-job/src/test/java/org/apache/cassandra/diff/TestUtils.java
new file mode 100644
index 0000000..ad9e1c4
--- /dev/null
+++ b/spark-job/src/test/java/org/apache/cassandra/diff/TestUtils.java
@@ -0,0 +1,40 @@
+/*
+ * 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.diff;
+
+import static org.junit.Assert.fail;
+
+public class TestUtils {
+
+    public static void assertThreadWaits(Thread t) {
+        for (int i=0; i < 1000; i++) {
+            try {
+                Thread.sleep(1);
+            }
+            catch (InterruptedException e) {
+                fail("Unexpected InterruptedException");
+            }
+            if (t.getState() == Thread.State.WAITING)
+                return;
+        }
+        fail(String.format("Thread %s expected to enter WAITING state, but failed to do so", t.getName()));
+    }
+
+}
diff --git a/spark-job/src/test/resources/cql-stress-narrow1.yaml b/spark-job/src/test/resources/cql-stress-narrow1.yaml
new file mode 100644
index 0000000..dae375b
--- /dev/null
+++ b/spark-job/src/test/resources/cql-stress-narrow1.yaml
@@ -0,0 +1,62 @@
+#
+# Keyspace info
+#
+keyspace: difftest
+
+#
+# The CQL for creating a keyspace (optional if it already exists)
+#
+keyspace_definition: |
+  CREATE KEYSPACE difftest WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1};
+
+#
+# Table info
+#
+table: narrow1
+
+#
+# The CQL for creating a table you wish to stress (optional if it already exists)
+#
+table_definition: |
+  CREATE TABLE narrow1 (
+        pk int,
+        v1 text,
+        v2 int,
+        PRIMARY KEY(pk)
+  ) 
+    WITH compaction = { 'class':'LeveledCompactionStrategy' }
+
+#
+# Optional meta information on the generated columns in the above table
+# The min and max only apply to text and blob types
+# The distribution field represents the total unique population
+# distribution of that column across rows.  Supported types are
+# 
+#      EXP(min..max)                        An exponential distribution over the range [min..max]
+#      EXTREME(min..max,shape)              An extreme value (Weibull) distribution over the range [min..max]
+#      GAUSSIAN(min..max,stdvrng)           A gaussian/normal distribution, where mean=(min+max)/2, and stdev is (mean-min)/stdvrng
+#      GAUSSIAN(min..max,mean,stdev)        A gaussian/normal distribution, with explicitly defined mean and stdev
+#      UNIFORM(min..max)                    A uniform distribution over the range [min, max]
+#      FIXED(val)                           A fixed distribution, always returning the same value
+#      Aliases: extr, gauss, normal, norm, weibull
+#
+#      If preceded by ~, the distribution is inverted
+#
+# Defaults for all columns are size: uniform(4..8), population: uniform(1..100B), cluster: fixed(1)
+#
+columnspec:
+  - name: pk
+    population: uniform(1..1B)
+  - name: v1
+    size: fixed(20)
+  - name: v2
+    population: uniform(1..100)
+
+insert:
+  partitions: fixed(50)       # number of unique partitions to update in a single operation
+  batchtype: UNLOGGED         # type of batch to use
+
+queries:
+   simple1:
+      cql: select * from narrow1 where pk = ? LIMIT 100
+      fields: samerow         # samerow or multirow (select arguments from the same row, or randomly from all rows in the partition)
diff --git a/spark-job/src/test/resources/cql-stress-wide1.yaml b/spark-job/src/test/resources/cql-stress-wide1.yaml
new file mode 100644
index 0000000..145ecce
--- /dev/null
+++ b/spark-job/src/test/resources/cql-stress-wide1.yaml
@@ -0,0 +1,69 @@
+#
+# Keyspace info
+#
+keyspace: difftest
+
+#
+# The CQL for creating a keyspace (optional if it already exists)
+#
+keyspace_definition: |
+  CREATE KEYSPACE difftest WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1};
+
+#
+# Table info
+#
+table: wide1
+
+#
+# The CQL for creating a table you wish to stress (optional if it already exists)
+#
+table_definition: |
+  CREATE TABLE wide1 (
+        pk int,
+        c1 int,
+        c2 int,
+        v1 text,
+        v2 int,
+        PRIMARY KEY(pk, c1, c2)
+  ) 
+    WITH compaction = { 'class':'LeveledCompactionStrategy' }
+
+#
+# Optional meta information on the generated columns in the above table
+# The min and max only apply to text and blob types
+# The distribution field represents the total unique population
+# distribution of that column across rows.  Supported types are
+# 
+#      EXP(min..max)                        An exponential distribution over the range [min..max]
+#      EXTREME(min..max,shape)              An extreme value (Weibull) distribution over the range [min..max]
+#      GAUSSIAN(min..max,stdvrng)           A gaussian/normal distribution, where mean=(min+max)/2, and stdev is (mean-min)/stdvrng
+#      GAUSSIAN(min..max,mean,stdev)        A gaussian/normal distribution, with explicitly defined mean and stdev
+#      UNIFORM(min..max)                    A uniform distribution over the range [min, max]
+#      FIXED(val)                           A fixed distribution, always returning the same value
+#      Aliases: extr, gauss, normal, norm, weibull
+#
+#      If preceded by ~, the distribution is inverted
+#
+# Defaults for all columns are size: uniform(4..8), population: uniform(1..100B), cluster: fixed(1)
+#
+columnspec:
+  - name: pk
+    population: uniform(1..1B)
+  - name: c1
+    cluster: uniform(1..100)
+  - name: c2
+    cluster: uniform(1..100)
+  - name: v1
+    size: fixed(20)
+  - name: v2
+    population: uniform(1..100)
+
+insert:
+  partitions: fixed(1)       # number of unique partitions to update in a single operation
+  batchtype: UNLOGGED        # type of batch to use
+
+
+queries:
+   simple1:
+      cql: select * from wide where pk = ? AND c1 = ? AND c2 = ? LIMIT 1
+      fields: samerow