Merge branch 'julienrf-aws-credentials'

Author: tarzanek

.github/wait-for-port.sh

@@ -0,0 +1,11 @@
+#!/usr/bin/env bash
+
+port=$1
+echo "Waiting for port ${port}"
+attempts=0
+max_attempts=30
+while ! nc -z 127.0.0.1 $port && [[ $attempts < $max_attempts ]] ; do
+    attempts=$((attempts+1))
+    sleep 1;
+    echo "waiting... (${attempts}/${max_attempts})"
+done

.github/workflows/tests.yml

@@ -2,8 +2,7 @@ name: Tests
 on:
   push:
     branches:
-      - master
-      - actions-test
+      - master      
   pull_request:
 
 jobs:
@@ -24,7 +23,12 @@ jobs:
       - name: Build migrator
         run: ./build.sh
       - name: Set up services
-        run: docker compose -f docker-compose-tests.yml up --wait
+        run: |
+          docker compose -f docker-compose-tests.yml up -d
+          .github/wait-for-port.sh 8000 # ScyllaDB
+          .github/wait-for-port.sh 8001 # DynamoDB
+          .github/wait-for-port.sh 8080 # Spark master
+          .github/wait-for-port.sh 8081 # Spark worker
       - name: Run tests
         run: sbt test
       - name: Stop services

.gitmodules

@@ -2,9 +2,6 @@
 	path = spark-cassandra-connector
 	url = https://github.com/scylladb/spark-cassandra-connector
 	branch = feature/track-token-ranges
-[submodule "spark-dynamodb"]
-	path = spark-dynamodb
-	url = https://github.com/scylladb/spark-dynamodb
 [submodule "spark-kinesis"]
 	path = spark-kinesis
 	url = https://github.com/scylladb/spark-kinesis

.travis.yml

@@ -41,3 +41,38 @@ sbt migrator/assembly
 ~~~
 
 And then re-run the tests.
+
+## Debugging
+
+The tests involve the execution of code on several locations:
+- locally (ie, on the machine where you invoke `sbt test`): tests initialization and assertions
+- on the Spark master node: the `Migrator` entry point
+- on the Spark worker node: RDD operations
+
+In all those cases, it is possible to debug them by using the Java Debug Wire Protocol.
+
+### Local Debugging
+
+Follow the procedure documented [here](https://stackoverflow.com/a/15505308/561721).
+
+### Debugging on the Spark Master Node
+
+1. In the file `MigratorSuite.scala`, uncomment the line that sets the
+   `spark.driver.extraJavaOptions`.
+2. Set up the remote debugger of your IDE to listen to the port 5005.
+3. Run a test
+4. When the test starts a Spark job, it waits for the remote debugger
+5. Start the remote debugger from your IDE.
+6. The test execution resumes, and you can interact with it from your debugger.
+
+### Debugging on the Spark Worker Node
+
+1. In the file `MigratorSuite.scala`, uncomment the line that sets the
+   `spark.executor.extraJavaOptions`.
+2. Set up the remote debugger of your IDE to listen to the port 5006.
+3. Run a test
+4. When the test starts an RDD operation, it waits for the remote debugger.
+   Note that the Spark master node will not display the output of the worker node,
+   but you can see it in the worker web UI: http://localhost:8081/.
+5. Start the remote debugger from your IDE.
+6. The test execution resumes, and you can interact with it from your debugger.

CONTRIBUTING.md

@@ -30,6 +30,7 @@ lazy val migrator = (project in file("migrator")).settings(
     "com.amazonaws"    % "aws-java-sdk-dynamodb" % awsSdkVersion,
     ("com.amazonaws" % "dynamodb-streams-kinesis-adapter" % "1.5.2")
       .excludeAll(InclExclRule("com.fasterxml.jackson.core")),
+    "com.amazon.emr" % "emr-dynamodb-hadoop" % "4.16.0",
     "org.yaml"       % "snakeyaml"      % "1.23",
     "io.circe"       %% "circe-yaml"    % "0.9.0",
     "io.circe"       %% "circe-generic" % "0.9.0",

build.sbt

@@ -14,9 +14,6 @@ trap "rm -rf $TMPDIR" EXIT
 pushd spark-cassandra-connector
 sbt -Djava.io.tmpdir="$TMPDIR" ++2.11.12 assembly
 popd
-pushd spark-dynamodb
-sbt assembly
-popd
 pushd spark-kinesis
 sbt assembly
 popd
@@ -26,7 +23,6 @@ if [ ! -d "./migrator/lib" ]; then
 fi
 
 cp ./spark-cassandra-connector/connector/target/scala-2.11/spark-cassandra-connector-assembly-*.jar ./migrator/lib
-cp ./spark-dynamodb/target/scala-2.11/spark-dynamodb-assembly-*.jar ./migrator/lib
 cp ./spark-kinesis/target/scala-2.11/spark-streaming-kinesis-asl-assembly-*.jar ./migrator/lib
 
 sbt -Djava.io.tmpdir="$TMPDIR" migrator/assembly

build.sh

@@ -36,6 +36,7 @@ services:
     networks:
       - scylla
     expose:
+      - 5005
       - 7001
       - 7002
       - 7003
@@ -46,12 +47,15 @@ services:
       - 6066
     ports:
       - 4040:4040
+      - 5005:5005
       - 6066:6066
       - 7077:7077
       - 8080:8080
     volumes:
       - ./migrator/target/scala-2.11:/jars
       - ./tests/src/test/configurations:/app/configurations
+      # Workaround for https://github.com/awslabs/emr-dynamodb-connector/issues/50
+      - ${PWD}/tests/docker/job-flow.json:/mnt/var/lib/info/job-flow.json
 
   spark-worker:
     image: bde2020/spark-worker:2.4.4-hadoop2.7
@@ -65,13 +69,15 @@ services:
     networks:
       - scylla
     expose:
+      - 5006
       - 7012
       - 7013
       - 7014
       - 7015
       - 7016
       - 8881
     ports:
+      - 5006:5006
       - 8081:8081
     depends_on:
       - spark-master

docker-compose-tests.yml

@@ -4,7 +4,6 @@ import com.amazonaws.auth.AWSCredentialsProvider
 import com.amazonaws.client.builder.AwsClientBuilder
 import com.amazonaws.services.dynamodbv2.{
   AmazonDynamoDBClientBuilder,
-  AmazonDynamoDBStreamsClient,
   AmazonDynamoDBStreamsClientBuilder
 }
 import com.amazonaws.services.dynamodbv2.model.{
@@ -17,7 +16,14 @@ import com.amazonaws.services.dynamodbv2.model.{
   TableDescription,
   UpdateTableRequest
 }
-import com.scylladb.migrator.config.{ DynamoDBEndpoint, SourceSettings, TargetSettings }
+import com.scylladb.migrator.config.{
+  AWSCredentials,
+  DynamoDBEndpoint,
+  SourceSettings,
+  TargetSettings
+}
+import org.apache.hadoop.dynamodb.DynamoDBConstants
+import org.apache.hadoop.mapred.JobConf
 import org.apache.log4j.LogManager
 
 import scala.util.{ Failure, Success, Try }
@@ -64,7 +70,8 @@ object DynamoUtils {
 
   def enableDynamoStream(source: SourceSettings.DynamoDB): Unit = {
     val sourceClient = buildDynamoClient(source.endpoint, source.credentials, source.region)
-    val sourceStreamsClient = buildDynamoStreamsClient(source.credentials, source.region)
+    val sourceStreamsClient =
+      buildDynamoStreamsClient(source.endpoint, source.credentials, source.region)
 
     sourceClient
       .updateTable(
@@ -114,12 +121,67 @@ object DynamoUtils {
     builder.build()
   }
 
-  def buildDynamoStreamsClient(creds: Option[AWSCredentialsProvider], region: Option[String]) = {
+  def buildDynamoStreamsClient(endpoint: Option[DynamoDBEndpoint],
+                               creds: Option[AWSCredentialsProvider],
+                               region: Option[String]) = {
     val builder = AmazonDynamoDBStreamsClientBuilder.standard()
 
+    endpoint.foreach { endpoint =>
+      builder
+        .withEndpointConfiguration(
+          new AwsClientBuilder.EndpointConfiguration(
+            endpoint.renderEndpoint,
+            region.getOrElse("empty")))
+    }
     creds.foreach(builder.withCredentials)
     region.foreach(builder.withRegion)
 
     builder.build()
   }
+
+  /**
+    * Optionally set a configuration. If `maybeValue` is empty, nothing is done. Otherwise,
+    * its value is set to the `name` property on the `jobConf`.
+    *
+    * @param jobConf    Target JobConf to configure
+    * @param name       Name of the Hadoop configuration key
+    * @param maybeValue Optional value to set.
+    */
+  def setOptionalConf(jobConf: JobConf, name: String, maybeValue: Option[String]): Unit =
+    for (value <- maybeValue) {
+      jobConf.set(name, value)
+    }
+
+  /**
+    * Set the common configuration of both read and write DynamoDB jobs.
+    * @param jobConf             Target JobConf to configure
+    * @param maybeRegion         AWS region
+    * @param maybeEndpoint       AWS endpoint
+    * @param maybeScanSegments   Scan segments
+    * @param maybeMaxMapTasks    Max map tasks
+    * @param maybeAwsCredentials AWS credentials
+    */
+  def setDynamoDBJobConf(jobConf: JobConf,
+                         maybeRegion: Option[String],
+                         maybeEndpoint: Option[DynamoDBEndpoint],
+                         maybeScanSegments: Option[Int],
+                         maybeMaxMapTasks: Option[Int],
+                         maybeAwsCredentials: Option[AWSCredentials]): Unit = {
+    setOptionalConf(jobConf, DynamoDBConstants.REGION, maybeRegion)
+    setOptionalConf(jobConf, DynamoDBConstants.ENDPOINT, maybeEndpoint.map(_.renderEndpoint))
+    setOptionalConf(jobConf, DynamoDBConstants.SCAN_SEGMENTS, maybeScanSegments.map(_.toString))
+    setOptionalConf(jobConf, DynamoDBConstants.MAX_MAP_TASKS, maybeMaxMapTasks.map(_.toString))
+    for (credentials <- maybeAwsCredentials) {
+      jobConf.set(DynamoDBConstants.DYNAMODB_ACCESS_KEY_CONF, credentials.accessKey)
+      jobConf.set(DynamoDBConstants.DYNAMODB_SECRET_KEY_CONF, credentials.secretKey)
+    }
+    jobConf.set(
+      DynamoDBConstants.CUSTOM_CREDENTIALS_PROVIDER_CONF,
+      "com.amazonaws.auth.profile.ProfileCredentialsProvider")
+    jobConf.set(
+      "mapred.output.format.class",
+      "org.apache.hadoop.dynamodb.write.DynamoDBOutputFormat")
+    jobConf.set("mapred.input.format.class", "org.apache.hadoop.dynamodb.read.DynamoDBInputFormat")
+  }
+
 }