Stack Exchange
Consume Kafka with Spark in Zeppelin running on Docker
so I have a local Docker Container running with Spotify/kafka (running on windows 10).
To this container there is data published in a topic called "data".
I can consume the data from a Java Application running in Eclipse like so:
private String topicName;
    private ConsumerConfig config;

    public KafkaConsumer() throws Exception {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("zookeeper.connect", "localhost:2181");
        props.put("group.id", "test");
        props.put("client.id", this.getClass().getSimpleName());
        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("value.deserializer", StringDeserializer.class.getName());
        props.put("partition.assignment.strategy", "range");

        this.config = new ConsumerConfig(props);
        this.topicName = "data";
    }

    public void run() {
        ConsumerConnector connector = Consumer.createJavaConsumerConnector(config);
        Map<String, List<KafkaStream<byte[], byte[]>>> messageStreams = connector.createMessageStreams(ImmutableMap.of(topicName, 1));
        List<KafkaStream<byte[], byte[]>> streams = messageStreams.get(topicName);
        ExecutorService executor = Executors.newFixedThreadPool(streams.size());

        for ( final KafkaStream<byte[], byte[]> stream : streams) {
            executor.submit(new Runnable() {
                public void run() {
                    ConsumerIterator<byte[], byte[]> iterator = stream.iterator();
                    while (iterator.hasNext()) {
                        MessageAndMetadata<byte[], byte[]> messageAndMetadata = iterator.next();
                        String m = new String(messageAndMetadata.message());

                        System.out.println(m);
                    }
                }
            });
        }
    }

What I want now is to consume the same topic in the  dylanmei/zeppelin docker-container. I tried running this in a Zeppelin Notebook:
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe


val kafkaParams = Map[String, Object](
  "bootstrap.servers" -> "localhost:9092",
  "zookeeper.connect" -> "localhost:2181",
  "group.id" -> "example",
  "key.deserializer" -> classOf[StringDeserializer],
  "value.deserializer" -> classOf[StringDeserializer],
  "partition.assignment.strategy" -> "range",
)

val topics = Array("data")
val stream = KafkaUtils.createDirectStream[String, String](
  sc,
  PreferConsistent,
  Subscribe[String, String](topics, kafkaParams)
)

stream.map(record => (record.key, record.value))

Which I have taken from the spark kafka integration guide
But all I ever get is a Connection refused (which also happens when I insert a wrong port, so that tells me nothing):
java.net.ConnectException: Connection refused
at java.net.PlainSocketImpl.socketConnect(Native Method)
at java.net.AbstractPlainSocketImpl.doConnect(AbstractPlainSocketImpl.java:350)
at java.net.AbstractPlainSocketImpl.connectToAddress(AbstractPlainSocketImpl.java:206)
at java.net.AbstractPlainSocketImpl.connect(AbstractPlainSocketImpl.java:188)
at java.net.SocksSocketImpl.connect(SocksSocketImpl.java:392)
at java.net.Socket.connect(Socket.java:589)
at org.apache.thrift.transport.TSocket.open(TSocket.java:182)
at org.apache.zeppelin.interpreter.remote.ClientFactory.create(ClientFactory.java:51)
at org.apache.zeppelin.interpreter.remote.ClientFactory.create(ClientFactory.java:37)
at org.apache.commons.pool2.BasePooledObjectFactory.makeObject(BasePooledObjectFactory.java:60)
at org.apache.commons.pool2.impl.GenericObjectPool.create(GenericObjectPool.java:861)
at org.apache.commons.pool2.impl.GenericObjectPool.borrowObject(GenericObjectPool.java:435)
at org.apache.commons.pool2.impl.GenericObjectPool.borrowObject(GenericObjectPool.java:363)
at org.apache.zeppelin.interpreter.remote.RemoteInterpreterProcess.getClient(RemoteInterpreterProcess.java:96)
at org.apache.zeppelin.interpreter.remote.RemoteInterpreter.init(RemoteInterpreter.java:216)
at org.apache.zeppelin.interpreter.remote.RemoteInterpreter.getFormType(RemoteInterpreter.java:385)
at org.apache.zeppelin.interpreter.LazyOpenInterpreter.getFormType(LazyOpenInterpreter.java:105)
at org.apache.zeppelin.notebook.Paragraph.jobRun(Paragraph.java:306)
at org.apache.zeppelin.scheduler.Job.run(Job.java:176)
at org.apache.zeppelin.scheduler.RemoteScheduler$JobRunner.run(RemoteScheduler.java:329)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)

Hope someone can help me make it run, I don't have a clue.
Thank you and kind regards!

First thing i will do do is to login to docker and telnet ports 2181 and 9092 to find out if the process is listening

Have you tried with --net="host"?  I think the default is bridge, and you would need to use the bridge adaptors address in your connect string.

I think this should be a spark interpreter issue. I had the same problem and when i restarted my interpreter all went well. You might as well check your interpreter's configuration. At times you get that error when your interpreter is not well configured. 
