Fix rabbitmq close issue, add automatic recovery

README.md

@@ -8,28 +8,10 @@ from [RabbitMQ](https://www.rabbitmq.com/).
 
 ## Requirements
 
-This library requires Spark 2.0+, Scala 2.11+, RabbitMQ 3.5+
+ This library was updated to work with Spark 2.3+ and sbt instead of maven.
 
 ## Using the library
-
-There are two ways of using RabbitMQ-Receiver library:
-
-The first one is to add the next dependency in your pom.xml:
-
-```
-<dependency>
-  <groupId>com.stratio.receiver</groupId>
-  <artifactId>spark-rabbitmq</artifactId>
-  <version>LATEST</version>
-</dependency>
-```
-
-The other one is to clone the full repository and build the project:
-
-```
-git clone https://github.com/Stratio/spark-rabbitmq.git
-mvn clean install
-```
+Create assembly with sbt and import into your project
 
 This library includes two implementations for consuming messages from RabbitMQ with Spark Streaming: 
 
@@ -42,8 +24,7 @@ This library includes two implementations for consuming messages from RabbitMQ w
 
 ### Build
 
-`mvn clean package`
-
+sbt assembly
 
 ### Distributed Approach
 

build.sbt

@@ -0,0 +1,15 @@
+name := "spark-rabbitmq"
+
+version := "0.1"
+
+scalaVersion := "2.11.8"
+
+libraryDependencies ++= Seq(
+  "org.apache.hadoop" % "hadoop-mapreduce-client-core" % "2.7.4" % "provided",
+  "org.apache.spark" %% "spark-core" % "2.3.2" % "provided",
+  "org.apache.spark" %% "spark-sql" % "2.3.2" % "provided",
+  "org.apache.spark" %% "spark-streaming" % "2.3.2" % "provided",
+  "com.softwaremill.sttp" %% "core" % "1.3.8",
+  "com.softwaremill.sttp" %% "circe" % "1.3.8",
+  "com.typesafe.akka" %% "akka-actor" % "2.5.18",
+  "com.rabbitmq" % "amqp-client" % "3.6.6")

project/assembly.sbt

@@ -0,0 +1 @@
+addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.14.8")

project/build.properties

@@ -0,0 +1 @@
+sbt.version = 1.2.7

src/main/scala/org/apache/spark/streaming/rabbitmq/ConfigParameters.scala

@@ -45,8 +45,10 @@ object ConfigParameters {
   val AckTypeKey = "ackType"
   val FairDispatchKey = "fairDispatch"
   val PrefetchCount = "prefetchCount"
+  val AutomaticRecovery = "automaticRecovery"
+  val NetworkRecoveryInterval = "networkRecoveryInterval"
   val QueueConnectionPropertiesKeys =
-    List(DurableKey, ExclusiveKey, AutoDeleteKey, AckTypeKey, FairDispatchKey, PrefetchCount)
+    List(DurableKey, ExclusiveKey, AutoDeleteKey, AckTypeKey, FairDispatchKey, PrefetchCount, AutomaticRecovery,NetworkRecoveryInterval)
 
   /**
    * Queue Connection Defaults
@@ -61,6 +63,8 @@ object ConfigParameters {
   val DefaultHost = "localhost"
   val DefaultPrefetchCount = 1
   val DefaultCodeClose = "320"
+  val DefaultAutomaticRecovery = true
+  val DefaultNetworkRecoveryInterval = 10000
 
   /**
    * Message Consumed properties

src/main/scala/org/apache/spark/streaming/rabbitmq/consumer/Consumer.scala

@@ -203,6 +203,7 @@ object Consumer extends Logging with ConsumerParamsUtils {
 
     setVirtualHost(params)
     setUserPassword(params)
+    setAutomaticRecoveryEnabled(params)
 
     getChannel(params) match {
       case Success(channel) =>
@@ -225,6 +226,18 @@ object Consumer extends Logging with ConsumerParamsUtils {
       }
     }
 
+  /**
+   * https://www.rabbitmq.com/api-guide.html#recovery
+   * @param params
+   */
+  private def setAutomaticRecoveryEnabled(params: Map[String,String]): Unit = {
+    val recoveryState = params.getOrElse(AutomaticRecovery, DefaultAutomaticRecovery.toString).toBoolean
+    factory.setAutomaticRecoveryEnabled(recoveryState)
+    val recoveryInterval = params.getOrElse(NetworkRecoveryInterval, DefaultNetworkRecoveryInterval.toString).toInt
+    factory.setNetworkRecoveryInterval(recoveryInterval)
+  }
+
+
   private def setVirtualHost(params: Map[String, String]): Unit = {
     val vHost = params.get(VirtualHostKey)
 

src/main/scala/org/apache/spark/streaming/rabbitmq/distributed/RabbitMQRDD.scala

@@ -231,7 +231,7 @@ class RabbitMQRDD[R: ClassTag](
       log.info(s"******* Received $numMessages messages by Partition : ${part.index}  before close Channel ******")
       //Close the scheduler and the channel in the consumer
       scheduleProcess.cancel()
-      consumer.close()
+      //consumer.close()
     }
 
     private def finishIterationAndReturn(): R = {