spark-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Likith_Kailas <kailas.lik...@gmail.com>
Subject SparkStreaming connection exception
Date Thu, 24 Aug 2017 13:05:17 GMT
I have written a unit test which uses multithreading to start and stop
Sparkstreamingjob and kafkaproducer. All the dependencies have been declared
in maven pom.xml file.

When i run the test, once the all the kafka messages are read and the
threads are stopped i continue to get the below exception

 2017-08-19 17:08:16,783 INFO  [Executor task launch worker-0-
 SendThread(127.0.0.1:64040)] zookeeper.ClientCnxn 
(ClientCnxn.java:logStartConnect(1032)) - Opening socket connection to 
server 127.0.0.1/127.0.0.1:64040. Will not attempt to authenticate using 
SASL (unknown error)
2017-08-19 17:08:17,786 WARN  [Executor task launch worker-0-
SendThread(127.0.0.1:64040)] zookeeper.ClientCnxn
(ClientCnxn.java:run(1162)) - Session 0x15dfb08227f0001 for server null,
unexpected error, closing socket connection and attempting reconnect
java.net.ConnectException: Connection refused: no further information
at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method)
at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:717)
at
org.apache.zookeeper.ClientCnxnSocketNIO.doTransport(ClientCnxnSocketNIO.java:361)
at org.apache.zookeeper.ClientCnxn$SendThread.run(ClientCnxn.java:1141)

The code is as below :

@Test
    public void someKafkaTest() {

    try {


        //Thread controlling the Spark streaming
        Thread sparkStreamerThread = new Thread(
                new SparkStreamingJSonJob(new String[] { zookeeperConnect,
"my-consumer-group", "test", "1" }),
                "spark-streaming");
        sparkStreamerThread.start();


        Thread.sleep(10000);

        //Thread to start the producer
        Thread producerThread = new Thread(new KafkaJSonProducer(),
"producer");
        producerThread.start();

        //current kafkaTest thread to sleep for 1 second
        Thread.sleep(60000);

        producerThread.stop();

        int sparkAccVal = SparkStreamingJSonJob.getAccumulator().intValue();
        System.out.println("Spark Throughput value : " + sparkAccVal/60);

        while (sparkStreamerThread.isAlive())
            sparkStreamerThread.stop();
       } catch (Exception e) {
        e.printStackTrace();
        Assert.fail();
    }

}

I feel that the streamingjob continues to run even after stopping the
thread. Please help me on this.



--
View this message in context: http://apache-spark-user-list.1001560.n3.nabble.com/SparkStreaming-connection-exception-tp29103.html
Sent from the Apache Spark User List mailing list archive at Nabble.com.

---------------------------------------------------------------------
To unsubscribe e-mail: user-unsubscribe@spark.apache.org


Mime
View raw message