Adding TransactionProcessor to map on returning batch of data

harishreedharan · harishreedharan · commit 8136aa60068e · 2014-07-14T16:12:12.000-07:00
diff --git a/external/flume-sink/src/main/scala/org/apache/spark/flume/sink/SparkAvroCallbackHandler.scala b/external/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
   }

Original file line number	Diff line number	Diff line change
`@@ -62,6 +62,7 @@ private class SparkAvroCallbackHandler(val threads: Int, val channel: Channel,`
`62`	`62`	`transactionExecutorOpt.map(executor => {`
`63`	`63`	`executor.submit(processor)`
`64`	`64`	`})`
	`65`	`+ processorMap.put(sequenceNumber, processor)`
`65`	`66`	`// Wait until a batch is available - will be an error if`
`66`	`67`	`processor.getEventBatch`
`67`	`68`	`}`