show table name in case of provisioned throughput exception

Author: Ben Kempe

src/main/scala/asyncdynamo/TracingAmazonDynamoDB.scala

@@ -25,43 +25,43 @@ protected class TracingAmazonDynamoDB(delegate  : AmazonDynamoDB, eventStream :
 
   def deleteItem(deleteItemRequest: DeleteItemRequest) = {
     deleteItemRequest.setReturnConsumedCapacity(ReturnConsumedCapacity.TOTAL)
-    val (res, duration) = time (delegate.deleteItem(deleteItemRequest))
+    val (res, duration) = time (delegate.deleteItem(deleteItemRequest), deleteItemRequest.getTableName)
     pub(DynamoRequestExecuted(Operation(deleteItemRequest.getTableName, Write, "DeleteItem"), writeUnits = Option(scala.Double.unbox(res.getConsumedCapacity.getCapacityUnits)), duration = duration))
     res
   }
 
   def getItem(getItemRequest: GetItemRequest) = {
     getItemRequest.setReturnConsumedCapacity(ReturnConsumedCapacity.TOTAL)
-    val (res, duration) = time(delegate.getItem(getItemRequest))
+    val (res, duration) = time(delegate.getItem(getItemRequest), getItemRequest.getTableName)
     pub(DynamoRequestExecuted(Operation(getItemRequest.getTableName, Read, "GetItem"), readUnits = Option(scala.Double.unbox(res.getConsumedCapacity.getCapacityUnits)), duration = duration))
     res
   }
 
   def scan(scanRequest: ScanRequest) = {
     scanRequest.setReturnConsumedCapacity(ReturnConsumedCapacity.TOTAL)
-    val (res, duration) = time(delegate.scan(scanRequest))
+    val (res, duration) = time(delegate.scan(scanRequest), scanRequest.getTableName)
     pub(DynamoRequestExecuted(Operation(scanRequest.getTableName, Read, "Scan"), readUnits = Option(scala.Double.unbox(res.getConsumedCapacity.getCapacityUnits)), duration = duration))
     res
   }
 
 
   def updateItem(updateItemRequest: UpdateItemRequest) = {
     updateItemRequest.setReturnConsumedCapacity(ReturnConsumedCapacity.TOTAL)
-    val (res, duration) = time(delegate.updateItem(updateItemRequest))
+    val (res, duration) = time(delegate.updateItem(updateItemRequest), updateItemRequest.getTableName)
     pub(DynamoRequestExecuted(Operation(updateItemRequest.getTableName, Write, "UpdateItem"), writeUnits = Option(scala.Double.unbox(res.getConsumedCapacity.getCapacityUnits)), duration = duration))
     res
   }
 
   def query(queryRequest: QueryRequest) = {
     queryRequest.setReturnConsumedCapacity(ReturnConsumedCapacity.TOTAL)
-    val (res, duration) = time(delegate.query(queryRequest))
+    val (res, duration) = time(delegate.query(queryRequest), queryRequest.getTableName)
     pub(DynamoRequestExecuted(Operation(queryRequest.getTableName, Read, "Query"), readUnits = Option(scala.Double.unbox(res.getConsumedCapacity.getCapacityUnits)), duration = duration))
     res
   }
 
   def putItem(putItemRequest: PutItemRequest) = {
     putItemRequest.setReturnConsumedCapacity(ReturnConsumedCapacity.TOTAL)
-    val (res, duration) = time(delegate.putItem(putItemRequest))
+    val (res, duration) = time(delegate.putItem(putItemRequest), putItemRequest.getTableName)
     pub(DynamoRequestExecuted(Operation(putItemRequest.getTableName, Write, "PutItem"), writeUnits = Option(scala.Double.unbox(res.getConsumedCapacity.getCapacityUnits)), duration = duration))
     res
   }
@@ -70,30 +70,34 @@ protected class TracingAmazonDynamoDB(delegate  : AmazonDynamoDB, eventStream :
 
   def batchGetItem(batchGetItemRequest: BatchGetItemRequest) = {
     batchGetItemRequest.setReturnConsumedCapacity(ReturnConsumedCapacity.TOTAL)
-    val (res, duration) = time(delegate.batchGetItem(batchGetItemRequest))
+    val (res, duration) = time(delegate.batchGetItem(batchGetItemRequest), batchGetItemRequest.getRequestItems.keySet().mkString(","))
 
     res.getConsumedCapacity foreach {
-      case consumedCapacity => 
+      case consumedCapacity =>
         pub(DynamoRequestExecuted(Operation(consumedCapacity.getTableName(), Read, "BatchGetItem"), readUnits = Option(scala.Double.unbox(consumedCapacity.getCapacityUnits)), duration = duration))
     }
     res
   }
 
   def batchWriteItem(batchWriteItemRequest: BatchWriteItemRequest) = {
     batchWriteItemRequest.setReturnConsumedCapacity(ReturnConsumedCapacity.TOTAL)
-    val (res, duration) = time(delegate.batchWriteItem(batchWriteItemRequest))
+    val (res, duration) = time(delegate.batchWriteItem(batchWriteItemRequest),batchWriteItemRequest.getRequestItems.keySet().mkString(","))
     res.getConsumedCapacity foreach {
-      case consumedCapacity => 
+      case consumedCapacity =>
         pub(DynamoRequestExecuted(Operation(consumedCapacity.getTableName(), Write, "BatchWriteItem"), writeUnits = Option(scala.Double.unbox(consumedCapacity.getCapacityUnits)), duration = duration))
     }
     res
   }
 
   private def pub(op:DynamoRequestExecuted) = eventStream.publish(op)
 
-  def time[T](f: => T): (T, Long) = {
+  def time[T](f: => T, tables: => String): (T, Long) = {
     val start = System.currentTimeMillis()
-    val res = f
+    val res = try {
+      f
+    } finally {
+      case ptee: ProvisionedThroughputExceededException => new ProvisionedThroughputExceededException(s"provisioned throughput for the table(s) was exceeded: $tables", ptee)
+    }
     (res, System.currentTimeMillis() - start)
   }