previous attempts exceptions in ArangoDBDataWriterException (#42)

Author: rashtao

ChangeLog.md

@@ -5,6 +5,8 @@ to [Semantic Versioning](http://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
 
+- added previous attempts exceptions in `ArangoDBDataWriterException`
+
 ## [1.4.2] - 2023-03-16
 
 - added debug header `x-arango-spark-request-id`

arangodb-spark-commons/src/main/scala/org/apache/spark/sql/arangodb/commons/exceptions/ArangoDBDataWriterException.scala

@@ -1,4 +1,24 @@
 package org.apache.spark.sql.arangodb.commons.exceptions
 
-class ArangoDBDataWriterException(val exception: Exception, val attempts: Int = 1)
-  extends RuntimeException(s"Failed $attempts times, most recent failure: $exception", exception)
+/**
+ * Exception thrown after all writes attempts have failed.
+ * It contains the exceptions thrown at each attempt.
+ *
+ * @param exceptions array of exceptions thrown at each attempt
+ */
+class ArangoDBDataWriterException(val exceptions: Array[Exception])
+  extends RuntimeException(s"Failed ${exceptions.length} times: ${ArangoDBDataWriterException.toMessage(exceptions)}") {
+  val attempts: Int = exceptions.length
+
+  override def getCause: Throwable = exceptions(0)
+}
+
+private object ArangoDBDataWriterException {
+
+  // creates exception message
+  private def toMessage(exceptions: Array[Exception]): String = exceptions
+    .zipWithIndex
+    .map(it => s"""Attempt #${it._2 + 1}: ${it._1}""")
+    .mkString("[\n\t", ",\n\t", "\n]")
+
+}

arangodb-spark-datasource-2.4/src/main/scala/org/apache/spark/sql/arangodb/datasource/writer/ArangoDataWriter.scala

@@ -21,6 +21,7 @@ class ArangoDataWriter(schema: StructType, options: ArangoDBConf, partitionId: I
   extends DataWriter[InternalRow] with Logging {
 
   private var failures = 0
+  private var exceptions: List[Exception] = List()
   private var requestCount = 0L
   private var endpointIdx = partitionId
   private val endpoints = Stream.continually(options.driverOptions.endpoints).flatten
@@ -90,10 +91,12 @@ class ArangoDataWriter(schema: StructType, options: ArangoDBConf, partitionId: I
       client.saveDocuments(payload)
       logDebug(s"Received response #$requestCount for partition $partitionId")
       failures = 0
+      exceptions = List()
     } catch {
       case e: Exception =>
         client.shutdown()
         failures += 1
+        exceptions = e :: exceptions
         endpointIdx += 1
         if ((canRetry || isConnectionException(e)) && failures < options.writeOptions.maxAttempts) {
           val delay = computeDelay()
@@ -102,7 +105,7 @@ class ArangoDataWriter(schema: StructType, options: ArangoDBConf, partitionId: I
           client = createClient()
           saveDocuments(payload)
         } else {
-          throw new ArangoDBDataWriterException(e, failures)
+          throw new ArangoDBDataWriterException(exceptions.reverse.toArray)
         }
     }
   }

arangodb-spark-datasource-3.1/src/main/scala/org/apache/spark/sql/arangodb/datasource/writer/ArangoDataWriter.scala

@@ -21,6 +21,7 @@ class ArangoDataWriter(schema: StructType, options: ArangoDBConf, partitionId: I
   extends DataWriter[InternalRow] with Logging {
 
   private var failures = 0
+  private var exceptions: List[Exception] = List()
   private var requestCount = 0L
   private var endpointIdx = partitionId
   private val endpoints = Stream.continually(options.driverOptions.endpoints).flatten
@@ -91,10 +92,12 @@ class ArangoDataWriter(schema: StructType, options: ArangoDBConf, partitionId: I
       client.saveDocuments(payload)
       logDebug(s"Received response #$requestCount for partition $partitionId")
       failures = 0
+      exceptions = List()
     } catch {
       case e: Exception =>
         client.shutdown()
         failures += 1
+        exceptions = e :: exceptions
         endpointIdx += 1
         if ((canRetry || isConnectionException(e)) && failures < options.writeOptions.maxAttempts) {
           val delay = computeDelay()
@@ -103,7 +106,7 @@ class ArangoDataWriter(schema: StructType, options: ArangoDBConf, partitionId: I
           client = createClient()
           saveDocuments(payload)
         } else {
-          throw new ArangoDBDataWriterException(e, failures)
+          throw new ArangoDBDataWriterException(exceptions.reverse.toArray)
         }
     }
   }

arangodb-spark-datasource-3.2/src/main/scala/org/apache/spark/sql/arangodb/datasource/writer/ArangoDataWriter.scala

@@ -21,6 +21,7 @@ class ArangoDataWriter(schema: StructType, options: ArangoDBConf, partitionId: I
   extends DataWriter[InternalRow] with Logging {
 
   private var failures = 0
+  private var exceptions: List[Exception] = List()
   private var requestCount = 0L
   private var endpointIdx = partitionId
   private val endpoints = Stream.continually(options.driverOptions.endpoints).flatten
@@ -91,10 +92,12 @@ class ArangoDataWriter(schema: StructType, options: ArangoDBConf, partitionId: I
       client.saveDocuments(payload)
       logDebug(s"Received response #$requestCount for partition $partitionId")
       failures = 0
+      exceptions = List()
     } catch {
       case e: Exception =>
         client.shutdown()
         failures += 1
+        exceptions = e :: exceptions
         endpointIdx += 1
         if ((canRetry || isConnectionException(e)) && failures < options.writeOptions.maxAttempts) {
           val delay = computeDelay()
@@ -103,7 +106,7 @@ class ArangoDataWriter(schema: StructType, options: ArangoDBConf, partitionId: I
           client = createClient()
           saveDocuments(payload)
         } else {
-          throw new ArangoDBDataWriterException(e, failures)
+          throw new ArangoDBDataWriterException(exceptions.reverse.toArray)
         }
     }
   }

bin/clean.sh

@@ -0,0 +1,7 @@
+#!/bin/bash
+
+mvn clean -Pspark-2.4 -Pscala-2.11
+mvn clean -Pspark-2.4 -Pscala-2.12
+mvn clean -Pspark-3.1 -Pscala-2.12
+mvn clean -Pspark-3.2 -Pscala-2.12
+mvn clean -Pspark-3.2 -Pscala-2.13

bin/test.sh

@@ -0,0 +1,19 @@
+#!/bin/bash
+
+# exit when any command fails
+set -e
+
+mvn clean -Pspark-2.4 -Pscala-2.11
+mvn test -Pspark-2.4 -Pscala-2.11
+
+mvn clean -Pspark-2.4 -Pscala-2.12
+mvn test -Pspark-2.4 -Pscala-2.12
+
+mvn clean -Pspark-3.1 -Pscala-2.12
+mvn test -Pspark-3.1 -Pscala-2.12
+
+mvn clean -Pspark-3.2 -Pscala-2.12
+mvn test -Pspark-3.2 -Pscala-2.12
+
+mvn clean -Pspark-3.2 -Pscala-2.13
+mvn test -Pspark-3.2 -Pscala-2.13

integration-tests/src/test/scala/org/apache/spark/sql/arangodb/datasource/write/AbortTest.scala

@@ -87,7 +87,7 @@ class AbortTest extends BaseSparkTest {
     assertThat(thrown).isInstanceOf(classOf[SparkException])
     assertThat(thrown.getCause.getCause).isInstanceOf(classOf[ArangoDBDataWriterException])
     assertThat(thrown.getCause.getCause.asInstanceOf[ArangoDBDataWriterException].attempts).isEqualTo(1)
-    assertThat(thrown.getCause.getCause.getMessage).contains("Failed 1 times, most recent failure:")
+    assertThat(thrown.getCause.getCause.getMessage).contains("Failed 1 times:")
 
     val validInserted = db.query(
       s"""FOR d IN $collectionName FILTER d.name == "valid" RETURN d""",
@@ -121,7 +121,7 @@ class AbortTest extends BaseSparkTest {
     assertThat(thrown).isInstanceOf(classOf[SparkException])
     assertThat(thrown.getCause.getCause).isInstanceOf(classOf[ArangoDBDataWriterException])
     assertThat(thrown.getCause.getCause.asInstanceOf[ArangoDBDataWriterException].attempts).isEqualTo(10)
-    assertThat(thrown.getCause.getCause.getMessage).contains("Failed 10 times, most recent failure:")
+    assertThat(thrown.getCause.getCause.getMessage).contains("Failed 10 times:")
 
     val rootEx = thrown.getCause.getCause.getCause
     assertThat(rootEx).isInstanceOf(classOf[ArangoDBMultiException])