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
index 6a7c591455d5f..1f6e60815dd06 100644
--- 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
   }