Adding TransactionProcessor to map on returning batch of data

ThalesGroup · Jul 14, 2014 · 8136aa6 · 8136aa6
1 parent 86aa274
commit 8136aa6
Showing 1 changed file with 1 addition and 0 deletions.
diff --git a/...rnal/flume-sink/src/main/scala/org/apache/spark/flume/sink/SparkAvroCallbackHandler.scala b/...rnal/flume-sink/src/main/scala/org/apache/spark/flume/sink/SparkAvroCallbackHandler.scala
@@ -62,6 +62,7 @@ private class SparkAvroCallbackHandler(val threads: Int, val channel: Channel,
     transactionExecutorOpt.map(executor => {
       executor.submit(processor)
     })
+    processorMap.put(sequenceNumber, processor)
     // Wait until a batch is available - will be an error if
     processor.getEventBatch
   }