@@ -76,7 +76,7 @@ private[streaming] class FlumePollingReceiver(
76
76
lazy val receiverExecutor = Executors .newFixedThreadPool(parallelism,
77
77
new ThreadFactoryBuilder ().setDaemon(true ).setNameFormat(" Flume Receiver Thread - %d" ).build())
78
78
79
- private val connections = new LinkedBlockingQueue [FlumeConnection ]()
79
+ private lazy val connections = new LinkedBlockingQueue [FlumeConnection ]()
80
80
81
81
override def onStart (): Unit = {
82
82
// Create the connections to each Flume agent.
@@ -102,11 +102,11 @@ private[streaming] class FlumePollingReceiver(
102
102
logDebug(
103
103
" Received batch of " + events.size() + " events with sequence number: " + seq)
104
104
try {
105
- // Convert each Flume event to a serializable SparkPollingEvent
105
+ // Convert each Flume event to a serializable SparkFlumeEvent
106
106
val buffer = new ArrayBuffer [SparkFlumeEvent ](events.size())
107
107
var j = 0
108
108
while (j < events.size()) {
109
- buffer += sparkSinkEventToSparkFlumeEvent (events(j))
109
+ buffer += toSparkFlumeEvent (events(j))
110
110
j += 1
111
111
}
112
112
store(buffer)
@@ -156,9 +156,9 @@ private[streaming] class FlumePollingReceiver(
156
156
/**
157
157
* Utility method to convert [[SparkSinkEvent ]] to [[SparkFlumeEvent ]]
158
158
* @param event - Event to convert to SparkFlumeEvent
159
- * @return - The SparkSinkEvent generated from Spar
159
+ * @return - The SparkFlumeEvent generated from SparkSinkEvent
160
160
*/
161
- private def sparkSinkEventToSparkFlumeEvent (event : SparkSinkEvent ): SparkFlumeEvent = {
161
+ private def toSparkFlumeEvent (event : SparkSinkEvent ): SparkFlumeEvent = {
162
162
val sparkFlumeEvent = new SparkFlumeEvent ()
163
163
sparkFlumeEvent.event.setBody(event.getBody)
164
164
sparkFlumeEvent.event.setHeaders(event.getHeaders)
0 commit comments