aboutsummaryrefslogtreecommitdiff
path: root/external/flume
diff options
context:
space:
mode:
authorSean Owen <srowen@gmail.com>2014-07-17 12:20:48 -0700
committerTathagata Das <tathagata.das1565@gmail.com>2014-07-17 12:20:48 -0700
commit1fcd5dcdd8edb0e6989278c95e7f2c7d86c4efb2 (patch)
tree2b05046391f40d3dd51e73eece5d819328b970fa /external/flume
parentd988d345d5bec0668324386f3e81787f78e75e67 (diff)
downloadspark-1fcd5dcdd8edb0e6989278c95e7f2c7d86c4efb2.tar.gz
spark-1fcd5dcdd8edb0e6989278c95e7f2c7d86c4efb2.tar.bz2
spark-1fcd5dcdd8edb0e6989278c95e7f2c7d86c4efb2.zip
SPARK-1478.2 Fix incorrect NioServerSocketChannelFactory constructor call
The line break inadvertently means this was interpreted as a call to the no-arg constructor. This doesn't exist in older Netty even. (Also fixed a val name typo.) Author: Sean Owen <srowen@gmail.com> Closes #1466 from srowen/SPARK-1478.2 and squashes the following commits: 59c3501 [Sean Owen] Line break caused Scala to interpret NioServerSocketChannelFactory constructor as the no-arg version, which is not even present in some versions of Netty
Diffstat (limited to 'external/flume')
-rw-r--r--external/flume/src/main/scala/org/apache/spark/streaming/flume/FlumeInputDStream.scala10
1 files changed, 5 insertions, 5 deletions
diff --git a/external/flume/src/main/scala/org/apache/spark/streaming/flume/FlumeInputDStream.scala b/external/flume/src/main/scala/org/apache/spark/streaming/flume/FlumeInputDStream.scala
index 07ae88febf..56d2886b26 100644
--- a/external/flume/src/main/scala/org/apache/spark/streaming/flume/FlumeInputDStream.scala
+++ b/external/flume/src/main/scala/org/apache/spark/streaming/flume/FlumeInputDStream.scala
@@ -153,15 +153,15 @@ class FlumeReceiver(
private def initServer() = {
if (enableDecompression) {
- val channelFactory = new NioServerSocketChannelFactory
- (Executors.newCachedThreadPool(), Executors.newCachedThreadPool());
- val channelPipelieFactory = new CompressionChannelPipelineFactory()
+ val channelFactory = new NioServerSocketChannelFactory(Executors.newCachedThreadPool(),
+ Executors.newCachedThreadPool())
+ val channelPipelineFactory = new CompressionChannelPipelineFactory()
new NettyServer(
responder,
new InetSocketAddress(host, port),
- channelFactory,
- channelPipelieFactory,
+ channelFactory,
+ channelPipelineFactory,
null)
} else {
new NettyServer(responder, new InetSocketAddress(host, port))