问:
NotLeaderForPartitionException when using multiple instances in a consumer group in Kafka+SparkStreaming
I've been experimenting with multiple consumer instances in a Kafka consumer group but it always fails with kafka.common.NotLeaderForPartitionException.

My Kafka cluster consists of 3 brokers and the topic that has PartitionCount:2and ReplicationFactor:3.

SparkConsumer.java

```<code>public class SparkConsumer {

private static Function2<Integer, Integer, Integer> MyReducerFunc = (a, b) -> a + b;

public static void main(String[] args) throws Exception {
    if (args.length < 2) {
      System.err.println("Usage: SparkConsumer <brokers> <topics>\n" +

       "  <brokers> is a list of one or more Kafka brokers\n" +
              "  <topics> is a list of one or more kafka topics to consume from\n\n");
          System.exit(1);
        }

        //StreamingExamples.setStreamingLogLevels();

        String brokers = args[0];
        String topics = args[1];

        SparkConf sparkConf = new  SparkConf().setMaster("local[5]").setAppName("SparkConsumer").set("spark.driver.host", "localhost");

        JavaSparkContext sc = new JavaSparkContext(sparkConf);

        // Create a StreamingContext with a 2 second batch size
        JavaStreamingContext jssc = new JavaStreamingContext(sc, Durations.seconds(2));

        Set<String> topicsSet = new HashSet<>(Arrays.asList(topics.split(",")));

        Map<String, String> kafkaParams = new HashMap<>();
        kafkaParams.put("metadata.broker.list", brokers);
        //kafkaParams.put("auto.offset.reset", "smallest");
        kafkaParams.put("group.id", "SparkConsumerGrp");
        kafkaParams.put("zookeeper.connect", "localhost:2181");

        // Create direct kafka stream with brokers and topics
        JavaPairInputDStream<String, String> messages = KafkaUtils.createDirectStream(
            jssc,
            String.class,
            String.class,
            StringDecoder.class,
            StringDecoder.class,
            kafkaParams,
            topicsSet
        );
        //Aggregate data every 30 sec
        JavaPairDStream<String, String> messages2 =
                messages.window(Durations.seconds(30), Durations.seconds(30));

        messages2.foreachRDD(rdd -> {

            long numHits = rdd.count();

            if(numHits == 0)
                System.out.println("No new data fetched in last 30 sec");

            //Do Processing
            else{
                    System.out.println("\n\n----------------------------------Data fetched in the last 30 seconds: " + rdd.partitions().size()
                            + " partitions and " + numHits  + " records------------------\n\n");

                    //Convert to java log object
                    JavaRDD<ApacheAccessLog> logs = rdd.map(x-> x._2)
                                                        .map(ApacheAccessLog::parseFromLogLine)
                                                            .cache();

                    //Find the bot ip addresses
                    JavaRDD<String> iprdd =  logs.mapToPair(ip-> new Tuple2<>(ip.getIpAddress(),1))
                            .reduceByKey(MyReducerFunc)
                                .filter(botip-> botip._2 > 50)
                                    .keys();

                    //If we find something, we store it in results dir on hdfs

                    if(iprdd.count() > 0)
                    {
                        sc.hadoopConfiguration().set("fs.hdfs.impl", org.apache.hadoop.hdfs.DistributedFileSystem.class.getName());
                        sc.hadoopConfiguration().set("fs.file.impl", org.apache.hadoop.fs.LocalFileSystem.class.getName());

                        String timeStamp = new SimpleDateFormat("yyyy.MM.dd.HH.mm.ss").format(new Date());
                        //sc.hadoopConfiguration().set("mapreduce.output.basename", timeStamp);
                        iprdd.coalesce(1).saveAsTextFile("hdfs://quickstart.cloudera:8020/results/"+timeStamp);
                        JobConf jobConf=new JobConf();
                                System.out.println("\n\n-------------Resuts successfully written to /results on hdfs-------------\n");
                    }

                }

        });

        jssc.start();
        jssc.awaitTermination();
}
}
</code>```

My understanding from the documentation was that we can start another consumer process with the same groupid to add that instance to an existing group. Hence, I run this code on two separate terminals. However, this is the error that I always get:

At the first instance:

```<code>    17/06/06 00:53:26 ERROR DirectKafkaInputDStream: ArrayBuffer(kafka.common.NotLeaderForPartitionException, org.apache.spark.SparkException: Couldn't find leader offsets for Set([topic5,1]))
17/06/06 00:53:26 ERROR DirectKafkaInputDStream: ArrayBuffer(org.apache.spark.SparkException: Couldn't find leaders for Set([topic5,0], [topic5,1]))
17/06/06 00:53:27 ERROR JobScheduler: Error generating jobs for time 1496735582000 ms
org.apache.spark.SparkException: ArrayBuffer(org.apache.spark.SparkException: Couldn't find leaders for Set([topic5,0], [topic5,1]))
    at org.apache.spark.streaming.kafka.DirectKafkaInputDStream.latestLeaderOffsets(DirectKafkaInputDStream.scala:133)
    at org.apache.spark.streaming.kafka.DirectKafkaInputDStream.compute(DirectKafkaInputDStream.scala:158)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1$$anonfun$apply$7.apply(DStream.scala:341)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1$$anonfun$apply$7.apply(DStream.scala:341)
    at scala.util.DynamicVariable.withValue(DynamicVariable.scala:58)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1.apply(DStream.scala:340)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1.apply(DStream.scala:340)
    at org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:415)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1.apply(DStream.scala:335)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1.apply(DStream.scala:333)
    at scala.Option.orElse(Option.scala:289)
    at org.apache.spark.streaming.dstream.DStream.getOrCompute(DStream.scala:330)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$slice$2$$anonfun$apply$29.apply(DStream.scala:900)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$slice$2$$anonfun$apply$29.apply(DStream.scala:899)
    at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)
    at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)
    at scala.collection.Iterator$class.foreach(Iterator.scala:893)
    at scala.collection.AbstractIterator.foreach(Iterator.scala:1336)
    at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
    at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
    at scala.collection.TraversableLike$class.flatMap(TraversableLike.scala:241)
    at scala.collection.AbstractTraversable.flatMap(Traversable.scala:104)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$slice$2.apply(DStream.scala:899)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$slice$2.apply(DStream.scala:877)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
    at org.apache.spark.SparkContext.withScope(SparkContext.scala:701)
    at org.apache.spark.streaming.StreamingContext.withScope(StreamingContext.scala:264)
    at org.apache.spark.streaming.dstream.DStream.slice(DStream.scala:877)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$slice$1.apply(DStream.scala:871)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$slice$1.apply(DStream.scala:871)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
    at org.apache.spark.SparkContext.withScope(SparkContext.scala:701)
    at org.apache.spark.streaming.StreamingContext.withScope(StreamingContext.scala:264)
    at org.apache.spark.streaming.dstream.DStream.slice(DStream.scala:870)
    at org.apache.spark.streaming.dstream.WindowedDStream.compute(WindowedDStream.scala:65)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1$$anonfun$apply$7.apply(DStream.scala:341)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1$$anonfun$apply$7.apply(DStream.scala:341)
    at scala.util.DynamicVariable.withValue(DynamicVariable.scala:58)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1.apply(DStream.scala:340)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1.apply(DStream.scala:340)
    at org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:415)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1.apply(DStream.scala:335)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1.apply(DStream.scala:333)
    at scala.Option.orElse(Option.scala:289)
    at org.apache.spark.streaming.dstream.DStream.getOrCompute(DStream.scala:330)
    at org.apache.spark.streaming.dstream.ForEachDStream.generateJob(ForEachDStream.scala:48)
    at org.apache.spark.streaming.DStreamGraph$$anonfun$1.apply(DStreamGraph.scala:117)
    at org.apache.spark.streaming.DStreamGraph$$anonfun$1.apply(DStreamGraph.scala:116)
    at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)
    at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)
    at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
    at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
    at scala.collection.TraversableLike$class.flatMap(TraversableLike.scala:241)
    at scala.collection.AbstractTraversable.flatMap(Traversable.scala:104)
    at org.apache.spark.streaming.DStreamGraph.generateJobs(DStreamGraph.scala:116)
    at org.apache.spark.streaming.scheduler.JobGenerator$$anonfun$3.apply(JobGenerator.scala:249)
    at org.apache.spark.streaming.scheduler.JobGenerator$$anonfun$3.apply(JobGenerator.scala:247)
    at scala.util.Try$.apply(Try.scala:192)
    at org.apache.spark.streaming.scheduler.JobGenerator.generateJobs(JobGenerator.scala:247)
    at org.apache.spark.streaming.scheduler.JobGenerator.org$apache$spark$streaming$scheduler$JobGenerator$$processEvent(JobGenerator.scala:183)
    at org.apache.spark.streaming.scheduler.JobGenerator$$anon$1.onReceive(JobGenerator.scala:89)
    at org.apache.spark.streaming.scheduler.JobGenerator$$anon$1.onReceive(JobGenerator.scala:88)
    at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48)
Exception in thread "main" org.apache.spark.SparkException: ArrayBuffer(org.apache.spark.SparkException: Couldn't find leaders for Set([topic5,0], [topic5,1]))
    at org.apache.spark.streaming.kafka.DirectKafkaInputDStream.latestLeaderOffsets(DirectKafkaInputDStream.scala:133)
    at org.apache.spark.streaming.kafka.DirectKafkaInputDStream.compute(DirectKafkaInputDStream.scala:158)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1$$anonfun$apply$7.apply(DStream.scala:341)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1$$anonfun$apply$7.apply(DStream.scala:341)
    at scala.util.DynamicVariable.withValue(DynamicVariable.scala:58)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1.apply(DStream.scala:340)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1.apply(DStream.scala:340)
    at org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:415)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1.apply(DStream.scala:335)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1.apply(DStream.scala:333)
    at scala.Option.orElse(Option.scala:289)
    at org.apache.spark.streaming.dstream.DStream.getOrCompute(DStream.scala:330)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$slice$2$$anonfun$apply$29.apply(DStream.scala:900)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$slice$2$$anonfun$apply$29.apply(DStream.scala:899)
    at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)
    at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)
    at scala.collection.Iterator$class.foreach(Iterator.scala:893)
    at scala.collection.AbstractIterator.foreach(Iterator.scala:1336)
    at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
    at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
    at scala.collection.TraversableLike$class.flatMap(TraversableLike.scala:241)
    at scala.collection.AbstractTraversable.flatMap(Traversable.scala:104)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$slice$2.apply(DStream.scala:899)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$slice$2.apply(DStream.scala:877)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
    at org.apache.spark.SparkContext.withScope(SparkContext.scala:701)
    at org.apache.spark.streaming.StreamingContext.withScope(StreamingContext.scala:264)
    at org.apache.spark.streaming.dstream.DStream.slice(DStream.scala:877)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$slice$1.apply(DStream.scala:871)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$slice$1.apply(DStream.scala:871)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
    at org.apache.spark.SparkContext.withScope(SparkContext.scala:701)
    at org.apache.spark.streaming.StreamingContext.withScope(StreamingContext.scala:264)
    at org.apache.spark.streaming.dstream.DStream.slice(DStream.scala:870)
    at org.apache.spark.streaming.dstream.WindowedDStream.compute(WindowedDStream.scala:65)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1$$anonfun$apply$7.apply(DStream.scala:341)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1$$anonfun$apply$7.apply(DStream.scala:341)
    at scala.util.DynamicVariable.withValue(DynamicVariable.scala:58)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1.apply(DStream.scala:340)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1.apply(DStream.scala:340)
    at org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:415)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1.apply(DStream.scala:335)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1.apply(DStream.scala:333)
    at scala.Option.orElse(Option.scala:289)
    at org.apache.spark.streaming.dstream.DStream.getOrCompute(DStream.scala:330)
    at org.apache.spark.streaming.dstream.ForEachDStream.generateJob(ForEachDStream.scala:48)
    at org.apache.spark.streaming.DStreamGraph$$anonfun$1.apply(DStreamGraph.scala:117)
    at org.apache.spark.streaming.DStreamGraph$$anonfun$1.apply(DStreamGraph.scala:116)
    at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)
    at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)
    at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
    at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
    at scala.collection.TraversableLike$class.flatMap(TraversableLike.scala:241)
    at scala.collection.AbstractTraversable.flatMap(Traversable.scala:104)
    at org.apache.spark.streaming.DStreamGraph.generateJobs(DStreamGraph.scala:116)
    at org.apache.spark.streaming.scheduler.JobGenerator$$anonfun$3.apply(JobGenerator.scala:249)
    at org.apache.spark.streaming.scheduler.JobGenerator$$anonfun$3.apply(JobGenerator.scala:247)
    at scala.util.Try$.apply(Try.scala:192)
    at org.apache.spark.streaming.scheduler.JobGenerator.generateJobs(JobGenerator.scala:247)
    at org.apache.spark.streaming.scheduler.JobGenerator.org$apache$spark$streaming$scheduler$JobGenerator$$processEvent(JobGenerator.scala:183)
    at org.apache.spark.streaming.scheduler.JobGenerator$$anon$1.onReceive(JobGenerator.scala:89)
    at org.apache.spark.streaming.scheduler.JobGenerator$$anon$1.onReceive(JobGenerator.scala:88)
    at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48)
</code>```

At the second instance: (seems to occur when I'm calling the rdd.count() and happens right after the first instance crashes)

```<code>    Exception in thread "streaming-job-executor-0" java.lang.Error: java.lang.InterruptedException
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1148)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.InterruptedException
    at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:998)
    at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1304)
    at scala.concurrent.impl.Promise$DefaultPromise.tryAwait(Promise.scala:202)
    at scala.concurrent.impl.Promise$DefaultPromise.ready(Promise.scala:218)
    at scala.concurrent.impl.Promise$DefaultPromise.ready(Promise.scala:153)
    at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:619)
    at org.apache.spark.SparkContext.runJob(SparkContext.scala:1925)
    at org.apache.spark.SparkContext.runJob(SparkContext.scala:1938)
    at org.apache.spark.SparkContext.runJob(SparkContext.scala:1951)
    at org.apache.spark.SparkContext.runJob(SparkContext.scala:1965)
    at org.apache.spark.rdd.RDD.count(RDD.scala:1158)
    at org.apache.spark.api.java.JavaRDDLike$class.count(JavaRDDLike.scala:455)
    at org.apache.spark.api.java.AbstractJavaRDDLike.count(JavaRDDLike.scala:45)
    at hadoopTest.hadoopTest.SparkConsumer.lambda$1(SparkConsumer.java:85)
    at org.apache.spark.streaming.api.java.JavaDStreamLike$$anonfun$foreachRDD$1.apply(JavaDStreamLike.scala:272)
    at org.apache.spark.streaming.api.java.JavaDStreamLike$$anonfun$foreachRDD$1.apply(JavaDStreamLike.scala:272)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$foreachRDD$1$$anonfun$apply$mcV$sp$3.apply(DStream.scala:627)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$foreachRDD$1$$anonfun$apply$mcV$sp$3.apply(DStream.scala:627)
    at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(ForEachDStream.scala:51)
    at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1$$anonfun$apply$mcV$sp$1.apply(ForEachDStream.scala:51)
    at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1$$anonfun$apply$mcV$sp$1.apply(ForEachDStream.scala:51)
    at org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:415)
    at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply$mcV$sp(ForEachDStream.scala:50)
    at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply(ForEachDStream.scala:50)
    at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply(ForEachDStream.scala:50)
    at scala.util.Try$.apply(Try.scala:192)
    at org.apache.spark.streaming.scheduler.Job.run(Job.scala:39)
    at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler$$anonfun$run$1.apply$mcV$sp(JobScheduler.scala:256)
    at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler$$anonfun$run$1.apply(JobScheduler.scala:256)
    at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler$$anonfun$run$1.apply(JobScheduler.scala:256)
    at scala.util.DynamicVariable.withValue(DynamicVariable.scala:58)
    at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:255)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
... 2 more
</code>```

This is my pom.xml

```<code>    <?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
   <modelVersion>4.0.0</modelVersion>
   <groupId>hadoopTest</groupId>
   <artifactId>hadoopTest</artifactId>
   <version>0.0.1-SNAPSHOT</version>
   <packaging>jar</packaging>
   <name>hadoopTest</name>
   <url>http://maven.apache.org</url>
   <properties>
      <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
   </properties>
   <dependencies>
      <dependency>
         <groupId>junit</groupId>
         <artifactId>junit</artifactId>
         <version>3.8.1</version>
         <scope>test</scope>
      </dependency>
      <dependency>
         <groupId>org.apache.hadoop</groupId>
         <artifactId>hadoop-hdfs</artifactId>
         <version>2.6.0</version>
      </dependency>
      <dependency>
         <groupId>org.apache.hadoop</groupId>
         <artifactId>hadoop-auth</artifactId>
         <version>2.6.0</version>
      </dependency>
      <dependency>
         <groupId>org.apache.hadoop</groupId>
         <artifactId>hadoop-common</artifactId>
         <version>2.6.0</version>
      </dependency>
      <dependency>
         <groupId>org.apache.hadoop</groupId>
         <artifactId>hadoop-mapreduce-client-core</artifactId>
         <version>2.6.0</version>
      </dependency>
      <dependency>
         <groupId>org.apache.kafka</groupId>
         <artifactId>kafka-clients</artifactId>
         <version>0.8.2.1</version>
      </dependency>
      <dependency>
         <groupId>org.apache.kafka</groupId>
         <artifactId>kafka_2.11</artifactId>
         <version>0.8.2.1</version>
         <scope>compile</scope>
         <exclusions>
            <exclusion>
               <artifactId>jmxri</artifactId>
               <groupId>com.sun.jmx</groupId>
            </exclusion>
            <exclusion>
               <artifactId>jms</artifactId>
               <groupId>javax.jms</groupId>
            </exclusion>
            <exclusion>
               <artifactId>jmxtools</artifactId>
               <groupId>com.sun.jdmk</groupId>
            </exclusion>
         </exclusions>
      </dependency>
      <dependency>
         <groupId>org.apache.spark</groupId>
         <artifactId>spark-core_2.11</artifactId>
         <version>2.1.1</version>
         <exclusions>
            <exclusion>
               <groupId>javax.validation</groupId>
               <artifactId>validation-api</artifactId>
            </exclusion>
         </exclusions>
      </dependency>
      <dependency>
         <groupId>org.apache.spark</groupId>
         <artifactId>spark-sql_2.11</artifactId>
         <version>1.2.2</version>
      </dependency>
      <dependency>
         <groupId>org.apache.spark</groupId>
         <artifactId>spark-streaming-kafka-0-8_2.11</artifactId>
         <version>2.1.1</version>
      </dependency>
      <dependency>
         <groupId>org.slf4j</groupId>
         <artifactId>slf4j-log4j12</artifactId>
         <version>1.7.5</version>
      </dependency>
      <dependency>
         <groupId>org.apache.spark</groupId>
         <artifactId>spark-streaming_2.11</artifactId>
         <version>2.1.1</version>
      </dependency>
      <dependency>
         <groupId>org.apache.spark</groupId>
         <artifactId>spark-tags_2.11</artifactId>
         <version>2.1.1</version>
      </dependency>
      <dependency>
         <groupId>org.scala-lang</groupId>
         <artifactId>scala-library</artifactId>
         <version>2.11.8</version>
      </dependency>
      <dependency>
         <groupId>org.scala-lang</groupId>
         <artifactId>scala-reflect</artifactId>
         <version>2.10.2</version>
      </dependency>
      <dependency>
         <groupId>net.jpountz.lz4</groupId>
         <artifactId>lz4</artifactId>
         <version>1.3</version>
      </dependency>
   </dependencies>
</project>
</code>```

I have been trying to debug this the whole day and am at a loss. It says it cannot find the leaders even though all my brokers remain up. I'd appreciate any help. Thanks!

答: Turns out it was because I was using spark-streaming-kafka-0-8_2.11 instead of spark-streaming-kafka-0-10_2.11. Also you need to change version for kafka-clients to 0.10.2.0 to match with it.

Body:
I've been experimenting with multiple consumer instances in a Kafka consumer group but it always fails with kafka.common.NotLeaderForPartitionException.

My Kafka cluster consists of 3 brokers and the topic that has PartitionCount:2and ReplicationFactor:3.

SparkConsumer.java

public class SparkConsumer {

private static Function2<Integer, Integer, Integer> MyReducerFunc = (a, b) -> a + b;

public static void main(String[] args) throws Exception {
    if (args.length < 2) {
      System.err.println("Usage: SparkConsumer <brokers> <topics>\n" +

       "  <brokers> is a list of one or more Kafka brokers\n" +
              "  <topics> is a list of one or more kafka topics to consume from\n\n");
          System.exit(1);
        }

        //StreamingExamples.setStreamingLogLevels();

        String brokers = args[0];
        String topics = args[1];

        SparkConf sparkConf = new  SparkConf().setMaster("local[5]").setAppName("SparkConsumer").set("spark.driver.host", "localhost");

        JavaSparkContext sc = new JavaSparkContext(sparkConf);

        // Create a StreamingContext with a 2 second batch size
        JavaStreamingContext jssc = new JavaStreamingContext(sc, Durations.seconds(2));

        Set<String> topicsSet = new HashSet<>(Arrays.asList(topics.split(",")));

        Map<String, String> kafkaParams = new HashMap<>();
        kafkaParams.put("metadata.broker.list", brokers);
        //kafkaParams.put("auto.offset.reset", "smallest");
        kafkaParams.put("group.id", "SparkConsumerGrp");
        kafkaParams.put("zookeeper.connect", "localhost:2181");

        // Create direct kafka stream with brokers and topics
        JavaPairInputDStream<String, String> messages = KafkaUtils.createDirectStream(
            jssc,
            String.class,
            String.class,
            StringDecoder.class,
            StringDecoder.class,
            kafkaParams,
            topicsSet
        );
        //Aggregate data every 30 sec
        JavaPairDStream<String, String> messages2 =
                messages.window(Durations.seconds(30), Durations.seconds(30));

        messages2.foreachRDD(rdd -> {

            long numHits = rdd.count();

            if(numHits == 0)
                System.out.println("No new data fetched in last 30 sec");

            //Do Processing
            else{
                    System.out.println("\n\n----------------------------------Data fetched in the last 30 seconds: " + rdd.partitions().size()
                            + " partitions and " + numHits  + " records------------------\n\n");

                    //Convert to java log object
                    JavaRDD<ApacheAccessLog> logs = rdd.map(x-> x._2)
                                                        .map(ApacheAccessLog::parseFromLogLine)
                                                            .cache();

                    //Find the bot ip addresses
                    JavaRDD<String> iprdd =  logs.mapToPair(ip-> new Tuple2<>(ip.getIpAddress(),1))
                            .reduceByKey(MyReducerFunc)
                                .filter(botip-> botip._2 > 50)
                                    .keys();

                    //If we find something, we store it in results dir on hdfs

                    if(iprdd.count() > 0)
                    {
                        sc.hadoopConfiguration().set("fs.hdfs.impl", org.apache.hadoop.hdfs.DistributedFileSystem.class.getName());
                        sc.hadoopConfiguration().set("fs.file.impl", org.apache.hadoop.fs.LocalFileSystem.class.getName());

                        String timeStamp = new SimpleDateFormat("yyyy.MM.dd.HH.mm.ss").format(new Date());
                        //sc.hadoopConfiguration().set("mapreduce.output.basename", timeStamp);
                        iprdd.coalesce(1).saveAsTextFile("hdfs://quickstart.cloudera:8020/results/"+timeStamp);
                        JobConf jobConf=new JobConf();
                                System.out.println("\n\n-------------Resuts successfully written to /results on hdfs-------------\n");
                    }

                }

        });

        jssc.start();
        jssc.awaitTermination();
}
}

My understanding from the documentation was that we can start another consumer process with the same groupid to add that instance to an existing group. Hence, I run this code on two separate terminals. However, this is the error that I always get:

At the first instance:

    17/06/06 00:53:26 ERROR DirectKafkaInputDStream: ArrayBuffer(kafka.common.NotLeaderForPartitionException, org.apache.spark.SparkException: Couldn't find leader offsets for Set([topic5,1]))
17/06/06 00:53:26 ERROR DirectKafkaInputDStream: ArrayBuffer(org.apache.spark.SparkException: Couldn't find leaders for Set([topic5,0], [topic5,1]))
17/06/06 00:53:27 ERROR JobScheduler: Error generating jobs for time 1496735582000 ms
org.apache.spark.SparkException: ArrayBuffer(org.apache.spark.SparkException: Couldn't find leaders for Set([topic5,0], [topic5,1]))
    at org.apache.spark.streaming.kafka.DirectKafkaInputDStream.latestLeaderOffsets(DirectKafkaInputDStream.scala:133)
    at org.apache.spark.streaming.kafka.DirectKafkaInputDStream.compute(DirectKafkaInputDStream.scala:158)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1$$anonfun$apply$7.apply(DStream.scala:341)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1$$anonfun$apply$7.apply(DStream.scala:341)
    at scala.util.DynamicVariable.withValue(DynamicVariable.scala:58)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1.apply(DStream.scala:340)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1.apply(DStream.scala:340)
    at org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:415)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1.apply(DStream.scala:335)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1.apply(DStream.scala:333)
    at scala.Option.orElse(Option.scala:289)
    at org.apache.spark.streaming.dstream.DStream.getOrCompute(DStream.scala:330)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$slice$2$$anonfun$apply$29.apply(DStream.scala:900)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$slice$2$$anonfun$apply$29.apply(DStream.scala:899)
    at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)
    at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)
    at scala.collection.Iterator$class.foreach(Iterator.scala:893)
    at scala.collection.AbstractIterator.foreach(Iterator.scala:1336)
    at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
    at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
    at scala.collection.TraversableLike$class.flatMap(TraversableLike.scala:241)
    at scala.collection.AbstractTraversable.flatMap(Traversable.scala:104)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$slice$2.apply(DStream.scala:899)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$slice$2.apply(DStream.scala:877)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
    at org.apache.spark.SparkContext.withScope(SparkContext.scala:701)
    at org.apache.spark.streaming.StreamingContext.withScope(StreamingContext.scala:264)
    at org.apache.spark.streaming.dstream.DStream.slice(DStream.scala:877)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$slice$1.apply(DStream.scala:871)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$slice$1.apply(DStream.scala:871)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
    at org.apache.spark.SparkContext.withScope(SparkContext.scala:701)
    at org.apache.spark.streaming.StreamingContext.withScope(StreamingContext.scala:264)
    at org.apache.spark.streaming.dstream.DStream.slice(DStream.scala:870)
    at org.apache.spark.streaming.dstream.WindowedDStream.compute(WindowedDStream.scala:65)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1$$anonfun$apply$7.apply(DStream.scala:341)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1$$anonfun$apply$7.apply(DStream.scala:341)
    at scala.util.DynamicVariable.withValue(DynamicVariable.scala:58)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1.apply(DStream.scala:340)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1.apply(DStream.scala:340)
    at org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:415)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1.apply(DStream.scala:335)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1.apply(DStream.scala:333)
    at scala.Option.orElse(Option.scala:289)
    at org.apache.spark.streaming.dstream.DStream.getOrCompute(DStream.scala:330)
    at org.apache.spark.streaming.dstream.ForEachDStream.generateJob(ForEachDStream.scala:48)
    at org.apache.spark.streaming.DStreamGraph$$anonfun$1.apply(DStreamGraph.scala:117)
    at org.apache.spark.streaming.DStreamGraph$$anonfun$1.apply(DStreamGraph.scala:116)
    at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)
    at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)
    at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
    at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
    at scala.collection.TraversableLike$class.flatMap(TraversableLike.scala:241)
    at scala.collection.AbstractTraversable.flatMap(Traversable.scala:104)
    at org.apache.spark.streaming.DStreamGraph.generateJobs(DStreamGraph.scala:116)
    at org.apache.spark.streaming.scheduler.JobGenerator$$anonfun$3.apply(JobGenerator.scala:249)
    at org.apache.spark.streaming.scheduler.JobGenerator$$anonfun$3.apply(JobGenerator.scala:247)
    at scala.util.Try$.apply(Try.scala:192)
    at org.apache.spark.streaming.scheduler.JobGenerator.generateJobs(JobGenerator.scala:247)
    at org.apache.spark.streaming.scheduler.JobGenerator.org$apache$spark$streaming$scheduler$JobGenerator$$processEvent(JobGenerator.scala:183)
    at org.apache.spark.streaming.scheduler.JobGenerator$$anon$1.onReceive(JobGenerator.scala:89)
    at org.apache.spark.streaming.scheduler.JobGenerator$$anon$1.onReceive(JobGenerator.scala:88)
    at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48)
Exception in thread "main" org.apache.spark.SparkException: ArrayBuffer(org.apache.spark.SparkException: Couldn't find leaders for Set([topic5,0], [topic5,1]))
    at org.apache.spark.streaming.kafka.DirectKafkaInputDStream.latestLeaderOffsets(DirectKafkaInputDStream.scala:133)
    at org.apache.spark.streaming.kafka.DirectKafkaInputDStream.compute(DirectKafkaInputDStream.scala:158)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1$$anonfun$apply$7.apply(DStream.scala:341)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1$$anonfun$apply$7.apply(DStream.scala:341)
    at scala.util.DynamicVariable.withValue(DynamicVariable.scala:58)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1.apply(DStream.scala:340)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1.apply(DStream.scala:340)
    at org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:415)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1.apply(DStream.scala:335)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1.apply(DStream.scala:333)
    at scala.Option.orElse(Option.scala:289)
    at org.apache.spark.streaming.dstream.DStream.getOrCompute(DStream.scala:330)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$slice$2$$anonfun$apply$29.apply(DStream.scala:900)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$slice$2$$anonfun$apply$29.apply(DStream.scala:899)
    at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)
    at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)
    at scala.collection.Iterator$class.foreach(Iterator.scala:893)
    at scala.collection.AbstractIterator.foreach(Iterator.scala:1336)
    at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
    at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
    at scala.collection.TraversableLike$class.flatMap(TraversableLike.scala:241)
    at scala.collection.AbstractTraversable.flatMap(Traversable.scala:104)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$slice$2.apply(DStream.scala:899)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$slice$2.apply(DStream.scala:877)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
    at org.apache.spark.SparkContext.withScope(SparkContext.scala:701)
    at org.apache.spark.streaming.StreamingContext.withScope(StreamingContext.scala:264)
    at org.apache.spark.streaming.dstream.DStream.slice(DStream.scala:877)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$slice$1.apply(DStream.scala:871)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$slice$1.apply(DStream.scala:871)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
    at org.apache.spark.SparkContext.withScope(SparkContext.scala:701)
    at org.apache.spark.streaming.StreamingContext.withScope(StreamingContext.scala:264)
    at org.apache.spark.streaming.dstream.DStream.slice(DStream.scala:870)
    at org.apache.spark.streaming.dstream.WindowedDStream.compute(WindowedDStream.scala:65)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1$$anonfun$apply$7.apply(DStream.scala:341)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1$$anonfun$apply$7.apply(DStream.scala:341)
    at scala.util.DynamicVariable.withValue(DynamicVariable.scala:58)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1.apply(DStream.scala:340)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1$$anonfun$1.apply(DStream.scala:340)
    at org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:415)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1.apply(DStream.scala:335)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$getOrCompute$1.apply(DStream.scala:333)
    at scala.Option.orElse(Option.scala:289)
    at org.apache.spark.streaming.dstream.DStream.getOrCompute(DStream.scala:330)
    at org.apache.spark.streaming.dstream.ForEachDStream.generateJob(ForEachDStream.scala:48)
    at org.apache.spark.streaming.DStreamGraph$$anonfun$1.apply(DStreamGraph.scala:117)
    at org.apache.spark.streaming.DStreamGraph$$anonfun$1.apply(DStreamGraph.scala:116)
    at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)
    at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)
    at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
    at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
    at scala.collection.TraversableLike$class.flatMap(TraversableLike.scala:241)
    at scala.collection.AbstractTraversable.flatMap(Traversable.scala:104)
    at org.apache.spark.streaming.DStreamGraph.generateJobs(DStreamGraph.scala:116)
    at org.apache.spark.streaming.scheduler.JobGenerator$$anonfun$3.apply(JobGenerator.scala:249)
    at org.apache.spark.streaming.scheduler.JobGenerator$$anonfun$3.apply(JobGenerator.scala:247)
    at scala.util.Try$.apply(Try.scala:192)
    at org.apache.spark.streaming.scheduler.JobGenerator.generateJobs(JobGenerator.scala:247)
    at org.apache.spark.streaming.scheduler.JobGenerator.org$apache$spark$streaming$scheduler$JobGenerator$$processEvent(JobGenerator.scala:183)
    at org.apache.spark.streaming.scheduler.JobGenerator$$anon$1.onReceive(JobGenerator.scala:89)
    at org.apache.spark.streaming.scheduler.JobGenerator$$anon$1.onReceive(JobGenerator.scala:88)
    at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48)

At the second instance: (seems to occur when I'm calling the rdd.count() and happens right after the first instance crashes)

    Exception in thread "streaming-job-executor-0" java.lang.Error: java.lang.InterruptedException
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1148)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.InterruptedException
    at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:998)
    at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1304)
    at scala.concurrent.impl.Promise$DefaultPromise.tryAwait(Promise.scala:202)
    at scala.concurrent.impl.Promise$DefaultPromise.ready(Promise.scala:218)
    at scala.concurrent.impl.Promise$DefaultPromise.ready(Promise.scala:153)
    at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:619)
    at org.apache.spark.SparkContext.runJob(SparkContext.scala:1925)
    at org.apache.spark.SparkContext.runJob(SparkContext.scala:1938)
    at org.apache.spark.SparkContext.runJob(SparkContext.scala:1951)
    at org.apache.spark.SparkContext.runJob(SparkContext.scala:1965)
    at org.apache.spark.rdd.RDD.count(RDD.scala:1158)
    at org.apache.spark.api.java.JavaRDDLike$class.count(JavaRDDLike.scala:455)
    at org.apache.spark.api.java.AbstractJavaRDDLike.count(JavaRDDLike.scala:45)
    at hadoopTest.hadoopTest.SparkConsumer.lambda$1(SparkConsumer.java:85)
    at org.apache.spark.streaming.api.java.JavaDStreamLike$$anonfun$foreachRDD$1.apply(JavaDStreamLike.scala:272)
    at org.apache.spark.streaming.api.java.JavaDStreamLike$$anonfun$foreachRDD$1.apply(JavaDStreamLike.scala:272)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$foreachRDD$1$$anonfun$apply$mcV$sp$3.apply(DStream.scala:627)
    at org.apache.spark.streaming.dstream.DStream$$anonfun$foreachRDD$1$$anonfun$apply$mcV$sp$3.apply(DStream.scala:627)
    at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(ForEachDStream.scala:51)
    at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1$$anonfun$apply$mcV$sp$1.apply(ForEachDStream.scala:51)
    at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1$$anonfun$apply$mcV$sp$1.apply(ForEachDStream.scala:51)
    at org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:415)
    at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply$mcV$sp(ForEachDStream.scala:50)
    at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply(ForEachDStream.scala:50)
    at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply(ForEachDStream.scala:50)
    at scala.util.Try$.apply(Try.scala:192)
    at org.apache.spark.streaming.scheduler.Job.run(Job.scala:39)
    at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler$$anonfun$run$1.apply$mcV$sp(JobScheduler.scala:256)
    at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler$$anonfun$run$1.apply(JobScheduler.scala:256)
    at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler$$anonfun$run$1.apply(JobScheduler.scala:256)
    at scala.util.DynamicVariable.withValue(DynamicVariable.scala:58)
    at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:255)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
... 2 more

This is my pom.xml

    <?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
   <modelVersion>4.0.0</modelVersion>
   <groupId>hadoopTest</groupId>
   <artifactId>hadoopTest</artifactId>
   <version>0.0.1-SNAPSHOT</version>
   <packaging>jar</packaging>
   <name>hadoopTest</name>
   <url>http://maven.apache.org</url>
   <properties>
      <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
   </properties>
   <dependencies>
      <dependency>
         <groupId>junit</groupId>
         <artifactId>junit</artifactId>
         <version>3.8.1</version>
         <scope>test</scope>
      </dependency>
      <dependency>
         <groupId>org.apache.hadoop</groupId>
         <artifactId>hadoop-hdfs</artifactId>
         <version>2.6.0</version>
      </dependency>
      <dependency>
         <groupId>org.apache.hadoop</groupId>
         <artifactId>hadoop-auth</artifactId>
         <version>2.6.0</version>
      </dependency>
      <dependency>
         <groupId>org.apache.hadoop</groupId>
         <artifactId>hadoop-common</artifactId>
         <version>2.6.0</version>
      </dependency>
      <dependency>
         <groupId>org.apache.hadoop</groupId>
         <artifactId>hadoop-mapreduce-client-core</artifactId>
         <version>2.6.0</version>
      </dependency>
      <dependency>
         <groupId>org.apache.kafka</groupId>
         <artifactId>kafka-clients</artifactId>
         <version>0.8.2.1</version>
      </dependency>
      <dependency>
         <groupId>org.apache.kafka</groupId>
         <artifactId>kafka_2.11</artifactId>
         <version>0.8.2.1</version>
         <scope>compile</scope>
         <exclusions>
            <exclusion>
               <artifactId>jmxri</artifactId>
               <groupId>com.sun.jmx</groupId>
            </exclusion>
            <exclusion>
               <artifactId>jms</artifactId>
               <groupId>javax.jms</groupId>
            </exclusion>
            <exclusion>
               <artifactId>jmxtools</artifactId>
               <groupId>com.sun.jdmk</groupId>
            </exclusion>
         </exclusions>
      </dependency>
      <dependency>
         <groupId>org.apache.spark</groupId>
         <artifactId>spark-core_2.11</artifactId>
         <version>2.1.1</version>
         <exclusions>
            <exclusion>
               <groupId>javax.validation</groupId>
               <artifactId>validation-api</artifactId>
            </exclusion>
         </exclusions>
      </dependency>
      <dependency>
         <groupId>org.apache.spark</groupId>
         <artifactId>spark-sql_2.11</artifactId>
         <version>1.2.2</version>
      </dependency>
      <dependency>
         <groupId>org.apache.spark</groupId>
         <artifactId>spark-streaming-kafka-0-8_2.11</artifactId>
         <version>2.1.1</version>
      </dependency>
      <dependency>
         <groupId>org.slf4j</groupId>
         <artifactId>slf4j-log4j12</artifactId>
         <version>1.7.5</version>
      </dependency>
      <dependency>
         <groupId>org.apache.spark</groupId>
         <artifactId>spark-streaming_2.11</artifactId>
         <version>2.1.1</version>
      </dependency>
      <dependency>
         <groupId>org.apache.spark</groupId>
         <artifactId>spark-tags_2.11</artifactId>
         <version>2.1.1</version>
      </dependency>
      <dependency>
         <groupId>org.scala-lang</groupId>
         <artifactId>scala-library</artifactId>
         <version>2.11.8</version>
      </dependency>
      <dependency>
         <groupId>org.scala-lang</groupId>
         <artifactId>scala-reflect</artifactId>
         <version>2.10.2</version>
      </dependency>
      <dependency>
         <groupId>net.jpountz.lz4</groupId>
         <artifactId>lz4</artifactId>
         <version>1.3</version>
      </dependency>
   </dependencies>
</project>

I have been trying to debug this the whole day and am at a loss. It says it cannot find the leaders even though all my brokers remain up. I'd appreciate any help. Thanks!


来源: stackexchange

问:
I need to escape! Angular $sce not handling links to audio files (Strict Contextual Escaping)
I have Angular and $sce working on my project for displaying HTML special characters, and for a video player's links to video files in my database. But I can't get $sce to connect the HTML audio player to audio files on the same database.

This HTML works:

```<code><audio controls ng-src="https://mydatabase.com/aa6df923">
</code>```

This is the project HTML:

```<code><audio controls ng-src="{{audioSource}}">
</code>```

This doesn't work:

```<code>$scope.audioSource = "https://mydatabase.com/aa6df923";
</code>```

This also doesn't work:

```<code>app.controller('MyController', function($scope, $sce) {
    $scope.audioSource = $sce.trustAsResourceUrl("https://mydatabase.com/aa6df923");
});
</code>```

This doesn't work either:

```<code>app.controller('MyController', function($scope, $sce) {
    $scope.audioSource = $sce.trustAsResourceUrl("https://mydatabase.com/aa6df923");
    $scope.$apply();
});
</code>```

Not this one either:

```<code>$scope.audioSource = "../audio/myFile.mp3";
</code>```

And, finally, this filter works to escape HTML special characters but don't work to link the audio files:

```<code><audio controls ng-src="{{audioSource | trustAsHtml}}">

app.filter('trustAsHtml', [$sce', function($sce) {
    return function(html) {
        return $sce.trustAsHtml(html);
    };
}]);
</code>```

What am I missing?

答:
You can create a new filter called trustAsResourceUrl that will work the same as your trustAsHtml example.

Create a trustAsResourceUrl filter:

```<code>app.filter('trustAsResourceUrl', ['$sce', function($sce) {
    return function(url) {
        return $sce.trustAsResourceUrl(url);
    };
}]);
</code>```

note that there is a typo in your example - a ( between return and function

Use the filter in your template:

```<code><audio controls ng-src="{{ audioSource | trustAsResourceUrl }}">
</code>```

Keep in mind this will trust every URL you pass to the filter.

If you just want to whitelist a domain you can do this in your app.config().

```<code>angular.module('app')
.config(['$sceDelegateProvider', function($sceDelegateProvider) {
  // whitelist resource URLs
  $sceDelegateProvider.resourceUrlWhitelist([
    'self',
    'https://mydatabase.com/**'
  ]);
}]);
</code>```

Otherwise the code in your earlier examples should work, assuming the domains match, etc. If these solutions don't fix your issue you it would be helpful to post the full error you are seeing in the console.

ContentLicense: CC BY-SA 3.0

Body:
I have Angular and $sce working on my project for displaying HTML special characters, and for a video player's links to video files in my database. But I can't get $sce to connect the HTML audio player to audio files on the same database.

This HTML works:

<audio controls ng-src="https://mydatabase.com/aa6df923">

This is the project HTML:

<audio controls ng-src="{{audioSource}}">

This doesn't work:

$scope.audioSource = "https://mydatabase.com/aa6df923";

This also doesn't work:

app.controller('MyController', function($scope, $sce) {
    $scope.audioSource = $sce.trustAsResourceUrl("https://mydatabase.com/aa6df923");
});

This doesn't work either:

app.controller('MyController', function($scope, $sce) {
    $scope.audioSource = $sce.trustAsResourceUrl("https://mydatabase.com/aa6df923");
    $scope.$apply();
});

Not this one either:

$scope.audioSource = "../audio/myFile.mp3";

And, finally, this filter works to escape HTML special characters but don't work to link the audio files:

<audio controls ng-src="{{audioSource | trustAsHtml}}">

app.filter('trustAsHtml', [$sce', function($sce) {
    return function(html) {
        return $sce.trustAsHtml(html);
    };
}]);

What am I missing?

Body:
You can create a new filter called trustAsResourceUrl that will work the same as your trustAsHtml example.

Create a trustAsResourceUrl filter:

app.filter('trustAsResourceUrl', ['$sce', function($sce) {
    return function(url) {
        return $sce.trustAsResourceUrl(url);
    };
}]);

note that there is a typo in your example - a ( between return and function

Use the filter in your template:

<audio controls ng-src="{{ audioSource | trustAsResourceUrl }}">

Keep in mind this will trust every URL you pass to the filter.

If you just want to whitelist a domain you can do this in your app.config().

angular.module('app')
.config(['$sceDelegateProvider', function($sceDelegateProvider) {
  // whitelist resource URLs
  $sceDelegateProvider.resourceUrlWhitelist([
    'self',
    'https://mydatabase.com/**'
  ]);
}]);

Otherwise the code in your earlier examples should work, assuming the domains match, etc. If these solutions don't fix your issue you it would be helpful to post the full error you are seeing in the console.


问:
mysql count the number of matches based on a column
This is my example dataset I have groups with students assigned to them as shown below

```
uid | groupid         | studentid
49  | PZV7cUZCnLwNkSS | wTsBSkkg4Weo8R3
50  | PZV7cUZCnLwNkSS | aIuDhxfChg3enCf
97  | CwvkffFcBCRbzdw | hEwLxJmnJmZFAic
99  | CwvkffFcBCRbzdw | OKFfl58XVQMrAyC
126 | CwvkffFcBCRbzdw | dlH8udyTjNV3nXM
142 | 2vu1eqTCWVjgE58 | Q01Iz3lC2uUMBSB
143 | 2vu1eqTCWVjgE58 | vB5s8hfTaVtx3wO
144 | 2vu1eqTCWVjgE58 | 5O9HA5Z7wVhgi6l
145 | 2vu1eqTCWVjgE58 | OiEUOXNjK2D2s8F```

I am trying to output with the following information.
The problem I am having is the Group Size column getting it to output a count.

```
Studentid       | Groupid         | Group Size
wTsBSkkg4Weo8R3 | PZV7cUZCnLwNkSS | 2
aIuDhxfChg3enCf | PZV7cUZCnLwNkSS | 2
hEwLxJmnJmZFAic | CwvkffFcBCRbzdw | 3
OKFfl58XVQMrAyC | CwvkffFcBCRbzdw | 3
dlH8udyTjNV3nXM | CwvkffFcBCRbzdw | 3```

I have researched if I can you can use a where clause in the count, and does not seem like it will let me do that.  I thought about doing a sum but couldn't make that happen either.  I feel like I am missing something simple.

答:
Try this:</p>

```<code>SELECT *
FROM pony a
LEFT JOIN (
    SELECT COUNT(*), groupid
    FROM pony
    GROUP BY groupid
) b ON a.groupid = b.groupid
</code>```

Body:
This is my example dataset I have groups with students assigned to them as shown below

uid | groupid         | studentid
49  | PZV7cUZCnLwNkSS | wTsBSkkg4Weo8R3
50  | PZV7cUZCnLwNkSS | aIuDhxfChg3enCf
97  | CwvkffFcBCRbzdw | hEwLxJmnJmZFAic
99  | CwvkffFcBCRbzdw | OKFfl58XVQMrAyC
126 | CwvkffFcBCRbzdw | dlH8udyTjNV3nXM
142 | 2vu1eqTCWVjgE58 | Q01Iz3lC2uUMBSB
143 | 2vu1eqTCWVjgE58 | vB5s8hfTaVtx3wO
144 | 2vu1eqTCWVjgE58 | 5O9HA5Z7wVhgi6l
145 | 2vu1eqTCWVjgE58 | OiEUOXNjK2D2s8F

I am trying to output with the following information.
The problem I am having is the Group Size column getting it to output a count.

Studentid       | Groupid         | Group Size
wTsBSkkg4Weo8R3 | PZV7cUZCnLwNkSS | 2
aIuDhxfChg3enCf | PZV7cUZCnLwNkSS | 2
hEwLxJmnJmZFAic | CwvkffFcBCRbzdw | 3
OKFfl58XVQMrAyC | CwvkffFcBCRbzdw | 3
dlH8udyTjNV3nXM | CwvkffFcBCRbzdw | 3

I have researched if I can you can use a where clause in the count, and does not seem like it will let me do that.  I thought about doing a sum but couldn't make that happen either.  I feel like I am missing something simple.

Body:
Try this:

SELECT *
FROM pony a
LEFT JOIN (
    SELECT COUNT(*), groupid
    FROM pony
    GROUP BY groupid
) b ON a.groupid = b.groupid


问:
Application Installation Failed with message Invalid File
My friend is developing an app, and he gave me his project on a USB drive. My problem is that when I try to run it on my Mac, it gives me this error:

Application Installation Failed

Installation failed with message Invalid File: [...].apk.
  It is possible that this issue is resolved by uninstalling an existing version of the apk if it is present, and then re-installing.

How or where to change the settings or the direction in Android Studio?

答: I got the same problem , just choose clean project then rebuild project from build menu , it's solved my problem</p>

ContentLicense: CC BY-SA 4.0


问:
How to replace a document matched by _id?
Working on an API I want to do a full document update of a MongoDB object.

This is my current code that works but it feels wrong to have to delete the _id every time. Is there a better way to do this?

```<code>        PutDoco : function(doco){
            return new Promise(function(Resolve,Reject){
                delete doco._id;
                db.collection('docos').updateOne(
                    {"details.ID":doco.details.ID},
                    doco,
                    function(err,result){
                        if(err)
                            return Reject(err);

                        Resolve(result);
                    }
                );
            });
        },
</code>```

答:
You can use replaceOne instead when replacing all contents of a document (besides _id which is immutable). And because replaceOne returns a promise if you don't pass it a callback, you can reduce your whole function down to:

```<code>return db.collection('docos').replaceOne({"details.ID":doco.details.ID}, doco);
</code>```

However, it might be clearer (and faster) to find the document to replace using _id instead:

```<code>return db.collection('docos').replaceOne({_id: doco._id}, doco);
</code>```

Body:
Working on an API I want to do a full document update of a MongoDB object.

This is my current code that works but it feels wrong to have to delete the _id every time. Is there a better way to do this?

        PutDoco : function(doco){
            return new Promise(function(Resolve,Reject){
                delete doco._id;
                db.collection('docos').updateOne(
                    {"details.ID":doco.details.ID},
                    doco,
                    function(err,result){
                        if(err)
                            return Reject(err);

                        Resolve(result);
                    }
                );
            });
        },

Body:
You can use replaceOne instead when replacing all contents of a document (besides _id which is immutable). And because replaceOne returns a promise if you don't pass it a callback, you can reduce your whole function down to:

return db.collection('docos').replaceOne({"details.ID":doco.details.ID}, doco);

However, it might be clearer (and faster) to find the document to replace using _id instead:

return db.collection('docos').replaceOne({_id: doco._id}, doco);


问:
In Python, how do I get user's remote IP (their last hop) if they're connected over SSH?
I want to detect if the user is connected over SSH. In a term, the "env" command shows SSH_CONNECTION line. Accessed in Python in one of two ways:

```<code>#python:
import os
print os.getenv("SSH_CONNECTION")       #works
print os.environ.get("SSH_CONNECTION")  #works
</code>```

But, if the user has ran my program using SUDO (as they will need to), env$ dooesn't show SSH_CONNECTION. So Python can't see it:

```<code>#sudo python:
import os
print os.getenv("SSH_CONNECTION")       #not set
print os.environ.get("SSH_CONNECTION")  #not set
</code>```

The aim is to achieve the following:

```<code>#Detect if user is over remote IP
lRemoteIP=""                                     #Is set if user on SSH
lStr=os.environ.get("SSH_CONNECTION")            #Temp var
if lStr: lRemoteIP=lStr.split()[0].split("=")[1] #Store user's lasthop IP

#Later on in the code, for multiple purposes:
if lRemoteIP: pass #Do stuff (or not) depending on if they're on SSH
</code>```

How do I retrieve SSH_CONNECTION environment variable under SUDO, when its not present in env$ ?

Or more precisely: how can I detect if the current session is via SSH when sudo?

I'm not a natural at Linuxy-type things, so be gentle with me...

[EDIT:] METHOD 2: Giving up on env$, I've tried the following:

```<code>pstree -ps $$ | grep "sshd("
</code>```

If it returns anything then it means that the SSH daemon sits above the session. Ergo, it's a SSH connection. And the results are showing me the PIDs of the SSH daemons. Results of the pstree cmd:

```<code>init(1)---sshd(xxx)---sshd(xxx)---sshd(xxx)---bash(xxx)-+-grep(xxx)
</code>```

But I'm struggling to get a src IP from the PID. Any ideas on this avenue?

[EDIT] METHOD 3: /run/utmp contains details of SSH logins. In python:

```<code>import os
import sys

lStr=open("/var/run/utmp").read().replace('\x00','') #Remove all those null values which make things hard to read

#Get the pseudo-session ID (pts) minus the /dev/ that it starts with:
lCurSess=os.ttyname(sys.stdout.fileno()).replace('/dev/','')
#Answer is like pts/10  (pseudo-term session number 10)
#Search lStr for pts/10
lInt=lStr.find(lCurSess.replace('/dev/',''))
#Print /var/utmp starting with where it first mentions current pts:
print lStr[lInt:]
</code>```

So far, so good. This gives the following results (I've changed the IP and username to USERNAME)

```<code>pts/10/10USERNAME\x9e\x958Ym\xb2\x05\x0 74\x14pts/10s/10USERNAME192.168.1.1\xbf\x958Y\xf8\xa3\r\xc0\xa88\x01
</code>```

So, when it comes to extracting the IP from the file, there's some bumf inbetween the occurances of pts/10 and the IP. What's the best way to parse it, given that (I reckon) the precise distance from the match to the IP will be different under different circumstances?

答:
The OpenSSH daemon writes an entry to /var/run/utmp with the current terminal, the IP and the name of the user.  Check the output of the w or who commands that parse /var/run/utmp.

It's just a question of getting the current terminal (similar to the tty command) and extracting the information you want.

Use pyutmp like this:

```<code>from pyutmp import UtmpFile
import os
import sys

for utmp in UtmpFile():
    if os.ttyname(sys.stdout.fileno()) == utmp.ut_line:
        print '%s logged from %s on tty %s' % (utmp.ut_user, utmp.ut_host, utmp.ut_line)
</code>```

Then filter by using ut_pid field to parse the /proc/ut_pid/cmdline file which should contain:

sshd: ut_user [priv]

Body:
I want to detect if the user is connected over SSH. In a term, the "env" command shows SSH_CONNECTION line. Accessed in Python in one of two ways:

#python:
import os
print os.getenv("SSH_CONNECTION")       #works
print os.environ.get("SSH_CONNECTION")  #works

But, if the user has ran my program using SUDO (as they will need to), env$ dooesn't show SSH_CONNECTION. So Python can't see it:

#sudo python:
import os
print os.getenv("SSH_CONNECTION")       #not set
print os.environ.get("SSH_CONNECTION")  #not set

The aim is to achieve the following:

#Detect if user is over remote IP
lRemoteIP=""                                     #Is set if user on SSH
lStr=os.environ.get("SSH_CONNECTION")            #Temp var
if lStr: lRemoteIP=lStr.split()[0].split("=")[1] #Store user's lasthop IP

#Later on in the code, for multiple purposes:
if lRemoteIP: pass #Do stuff (or not) depending on if they're on SSH

How do I retrieve SSH_CONNECTION environment variable under SUDO, when its not present in env$ ?

Or more precisely: how can I detect if the current session is via SSH when sudo?

I'm not a natural at Linuxy-type things, so be gentle with me...

[EDIT:] METHOD 2: Giving up on env$, I've tried the following:

pstree -ps $$ | grep "sshd("

If it returns anything then it means that the SSH daemon sits above the session. Ergo, it's a SSH connection. And the results are showing me the PIDs of the SSH daemons. Results of the pstree cmd:

init(1)---sshd(xxx)---sshd(xxx)---sshd(xxx)---bash(xxx)-+-grep(xxx)

But I'm struggling to get a src IP from the PID. Any ideas on this avenue?

[EDIT] METHOD 3: /run/utmp contains details of SSH logins. In python:

import os
import sys

lStr=open("/var/run/utmp").read().replace('\x00','') #Remove all those null values which make things hard to read

#Get the pseudo-session ID (pts) minus the /dev/ that it starts with:
lCurSess=os.ttyname(sys.stdout.fileno()).replace('/dev/','')
#Answer is like pts/10  (pseudo-term session number 10)
#Search lStr for pts/10
lInt=lStr.find(lCurSess.replace('/dev/',''))
#Print /var/utmp starting with where it first mentions current pts:
print lStr[lInt:]

So far, so good. This gives the following results (I've changed the IP and username to USERNAME)

pts/10/10USERNAME\x9e\x958Ym\xb2\x05\x0 74\x14pts/10s/10USERNAME192.168.1.1\xbf\x958Y\xf8\xa3\r\xc0\xa88\x01

So, when it comes to extracting the IP from the file, there's some bumf inbetween the occurances of pts/10 and the IP. What's the best way to parse it, given that (I reckon) the precise distance from the match to the IP will be different under different circumstances?

Body:
The OpenSSH daemon writes an entry to /var/run/utmp with the current terminal, the IP and the name of the user.  Check the output of the w or who commands that parse /var/run/utmp.

It's just a question of getting the current terminal (similar to the tty command) and extracting the information you want.

Use pyutmp like this:

from pyutmp import UtmpFile
import os
import sys

for utmp in UtmpFile():
    if os.ttyname(sys.stdout.fileno()) == utmp.ut_line:
        print '%s logged from %s on tty %s' % (utmp.ut_user, utmp.ut_host, utmp.ut_line)

Then filter by using ut_pid field to parse the /proc/ut_pid/cmdline file which should contain:

sshd: ut_user [priv]


答:
try this</p>

```<code>SELECT T1.Studentid, T1.Groupid, T2.GroupCount
FROM Your_Table T1
INNER JOIN ( SELECT Groupid, count(*) AS GroupCount FROM Your_Table GROUP BY Groupid ) T2
ON T1.Groupid = T2.Groupid
</code>```

Title: mysql count the number of matches based on a column

Body:
try this

SELECT T1.Studentid, T1.Groupid, T2.GroupCount
FROM Your_Table T1
INNER JOIN ( SELECT Groupid, count(*) AS GroupCount FROM Your_Table GROUP BY Groupid ) T2
ON T1.Groupid = T2.Groupid


问:
Is it possible to have multiple styles inside a TextView?
Is it possible to set multiple styles for different pieces of text inside a TextView?

For instance, I am setting the text as follows:

```<code>tv.setText(line1 + "\n" + line2 + "\n" + word1 + "\t" + word2 + "\t" + word3);
</code>```

Is it possible to have a different style for each text element? E.g., line1 bold, word1 italic, etc.

The developer guide's Common Tasks and How to Do Them in Android includes Selecting, Highlighting, or Styling Portions of Text:

```<code>// Get our EditText object.
EditText vw = (EditText)findViewById(R.id.text);

// Set the EditText's text.
vw.setText("Italic, highlighted, bold.");

// If this were just a TextView, we could do:
// vw.setText("Italic, highlighted, bold.", TextView.BufferType.SPANNABLE);
// to force it to use Spannable storage so styles can be attached.
// Or we could specify that in the XML.

// Get the EditText's internal text storage
Spannable str = vw.getText();

// Create our span sections, and assign a format to each.
str.setSpan(new StyleSpan(android.graphics.Typeface.ITALIC), 0, 7, Spannable.SPAN_EXCLUSIVE_EXCLUSIVE);
str.setSpan(new BackgroundColorSpan(0xFFFFFF00), 8, 19, Spannable.SPAN_EXCLUSIVE_EXCLUSIVE);
str.setSpan(new StyleSpan(android.graphics.Typeface.BOLD), 21, str.length() - 1, Spannable.SPAN_EXCLUSIVE_EXCLUSIVE);
</code>```

But that uses explicit position numbers inside the text. Is there a cleaner way to do this?

答:
Using an auxiliary Spannable Class as Android String Resources shares at the bottom of the webpage. You can approach this by creatingCharSquences and giving them a style.

But in the example they give us, is just for bold, italic, and even colorize text. I needed to wrap several styles in aCharSequence in order to set them in a TextView. So to that Class (I named it CharSequenceStyles) I just added this function.

```<code>public static CharSequence applyGroup(LinkedList<CharSequence> content){
    SpannableStringBuilder text = new SpannableStringBuilder();
    for (CharSequence item : content) {
        text.append(item);
    }
    return text;
}
</code>```

And in the view I added this.

```<code>            message.push(postMessageText);
            message.push(limitDebtAmount);
            message.push(pretMessageText);
            TextView.setText(CharSequenceStyles.applyGroup(message));
</code>```

I hope this help you!

Body:
Is it possible to set multiple styles for different pieces of text inside a TextView?

For instance, I am setting the text as follows:

tv.setText(line1 + "\n" + line2 + "\n" + word1 + "\t" + word2 + "\t" + word3);

Is it possible to have a different style for each text element? E.g., line1 bold, word1 italic, etc.

The developer guide's Common Tasks and How to Do Them in Android includes Selecting, Highlighting, or Styling Portions of Text:

// Get our EditText object.
EditText vw = (EditText)findViewById(R.id.text);

// Set the EditText's text.
vw.setText("Italic, highlighted, bold.");

// If this were just a TextView, we could do:
// vw.setText("Italic, highlighted, bold.", TextView.BufferType.SPANNABLE);
// to force it to use Spannable storage so styles can be attached.
// Or we could specify that in the XML.

// Get the EditText's internal text storage
Spannable str = vw.getText();

// Create our span sections, and assign a format to each.
str.setSpan(new StyleSpan(android.graphics.Typeface.ITALIC), 0, 7, Spannable.SPAN_EXCLUSIVE_EXCLUSIVE);
str.setSpan(new BackgroundColorSpan(0xFFFFFF00), 8, 19, Spannable.SPAN_EXCLUSIVE_EXCLUSIVE);
str.setSpan(new StyleSpan(android.graphics.Typeface.BOLD), 21, str.length() - 1, Spannable.SPAN_EXCLUSIVE_EXCLUSIVE);

But that uses explicit position numbers inside the text. Is there a cleaner way to do this?

Body:
Using an auxiliary Spannable Class as Android String Resources shares at the bottom of the webpage. You can approach this by creatingCharSquences and giving them a style.

But in the example they give us, is just for bold, italic, and even colorize text. I needed to wrap several styles in aCharSequence in order to set them in a TextView. So to that Class (I named it CharSequenceStyles) I just added this function.

public static CharSequence applyGroup(LinkedList<CharSequence> content){
    SpannableStringBuilder text = new SpannableStringBuilder();
    for (CharSequence item : content) {
        text.append(item);
    }
    return text;
}

And in the view I added this.

            message.push(postMessageText);
            message.push(limitDebtAmount);
            message.push(pretMessageText);
            TextView.setText(CharSequenceStyles.applyGroup(message));

I hope this help you!


问:
Android: Populating ListView from SQLite database (multiple columns)
I am developing an app in Android and have a problem with passing data from databse to listview and viewing it.

Here is my database class with all methods:

```<code>package com.example.admin.planesproject;

import android.content.ContentValues;
import android.content.Context;
import android.database.Cursor;
import android.database.sqlite.SQLiteDatabase;
import android.database.sqlite.SQLiteOpenHelper;
import com.example.admin.planesproject.TodoTask;

public class PlanesDb {

private static final int DB_VERSION = 1;
private static final String DB_NAME = "database.db";
private static final String DB_TODO_TABLE = "PlanesDb";

//columns

public static final String KEY_ID = "_id";
public static final String ID_OPTIONS = "INTEGER PRIMARY KEY AUTOINCREMENT";
public static final int ID_COLUMN = 0;

public static final String KEY_DATE = "date";
public static final String DATE_OPTIONS = "TEXT NOT NULL";
public static final int DATE_COLUMN = 1;

public static final String KEY_DIRECTION = "direction";
public static final String DIRECTION_OPTIONS = "TEXT NOT NULL";
public static final int DIRECTION_COLUMN = 2;

public static final String KEY_MODEL = "model";
public static final String MODEL_OPTIONS = "TEXT";
public static final int MODEL_COLUMN = 3;

public static final String KEY_AIRLINES = "airlines";
public static final String AIRLINES_OPTIONS = "TEXT";
public static final int AIRLINES_COLUMN = 4;

public static final String KEY_FROM = "from";
public static final String FROM_OPTIONS = "TEXT";
public static final int FROM_COLUMN = 5;

public static final String KEY_TO = "to";
public static final String TO_OPTIONS = "TEXT";
public static final int TO_COLUMN = 6;

private static final String DB_CREATE_TODO_TABLE =
        "CREATE TABLE " + DB_TODO_TABLE + "( " +
                KEY_ID + " " + ID_OPTIONS + ", " +
                KEY_DATE + " " + DATE_OPTIONS + ", " +
                KEY_DIRECTION + " " + DIRECTION_OPTIONS + ", " +
                KEY_MODEL + " " + MODEL_OPTIONS + ", " +
                KEY_AIRLINES + " " + AIRLINES_OPTIONS + ", " +
                KEY_FROM + " " + FROM_OPTIONS + ", " +
                KEY_TO + " " + TO_OPTIONS + ");";

private static final String DROP_TODO_TABLE =
        "DROP TABLE IF EXISTS " + DB_TODO_TABLE;

private SQLiteDatabase db;
private Context context;
private DatabaseHelper dbHelper;

private static class DatabaseHelper extends SQLiteOpenHelper {
    public DatabaseHelper(Context context, String name,
                          SQLiteDatabase.CursorFactory factory, int version) {
        super(context, name, factory, version);
    }

    @Override
    public void onCreate(SQLiteDatabase db) {
        db.execSQL(DB_CREATE_TODO_TABLE);
    }

    @Override
    public void onUpgrade(SQLiteDatabase db, int oldVersion, int newVersion) {
        db.execSQL(DROP_TODO_TABLE);
        onCreate(db);
    }
}

public PlanesDb(Context context) {
    this.context = context;
}

public PlanesDb open(){
    dbHelper = new DatabaseHelper(context, DB_NAME, null, DB_VERSION);
    db = dbHelper.getWritableDatabase();
    return this;
}

public void close() {
    dbHelper.close();
}

public long insertTodo(String date, String direction, String model, String airlines, String from, String to) {
    ContentValues newTodoValues = new ContentValues();
    newTodoValues.put(KEY_DATE, date);
    newTodoValues.put(KEY_DIRECTION, direction);
    newTodoValues.put(KEY_MODEL, model);
    newTodoValues.put(KEY_AIRLINES, airlines);
    newTodoValues.put(KEY_FROM, from);
    newTodoValues.put(KEY_TO, to);
    return db.insert(DB_TODO_TABLE, null, newTodoValues);
}

public boolean updateTodo(TodoTask task) {
    long id = task.getId();
    String date = task.getDate();
    String direction = task.getDirection();
    String model = task.getModel();
    String airlines = task.getAirlines();
    String from = task.getFrom();
    String to = task.getTo();
    return updateTodo(id, date, direction, model, airlines, from, to);
}

public boolean updateTodo(long id, String date, String direction, String model, String airlines, String from, String to) {
    String where = KEY_ID + "=" + id;
    ContentValues updateTodoValues = new ContentValues();
    updateTodoValues.put(KEY_DATE, date);
    updateTodoValues.put(KEY_DIRECTION, direction);
    updateTodoValues.put(KEY_MODEL, model);
    updateTodoValues.put(KEY_AIRLINES, airlines);
    updateTodoValues.put(KEY_FROM, from);
    updateTodoValues.put(KEY_TO, to);
    return db.update(DB_TODO_TABLE, updateTodoValues, where, null) > 0;
}

public boolean deleteTodo(long id){
    String where = KEY_ID + "=" + id;
    return db.delete(DB_TODO_TABLE, where, null) > 0;
}

public Cursor getAllTodos() {
    String[] columns = {KEY_ID, KEY_DATE, KEY_DIRECTION, KEY_MODEL, KEY_AIRLINES, KEY_FROM, KEY_TO};
    return db.query(DB_TODO_TABLE, columns, null, null, null, null, null);
}
}
</code>```

I'd like to have a listview showing everything from my table, but I cannot grasp how to do it. Tutorials don't help much as most of them only shows how to work with one row database.

My ToDo Task with all getters and setters:

```<code>public class TodoTask {
private long id;
private String date;
private String direction;
private String model;
private String airlines;
private String from;
private String to;

public TodoTask(long id, String date, String direction, String model, String airlines, String from, String to) {
    this.id = id;
    this.date = date;
    this.direction = direction;
    this.model = model;
    this.airlines = airlines;
    this.from = from;
    this.to = to;
}

public long getId() {
    return id;
}

public void setId(long id) {
    this.id = id;
}

public String getDate(){
    return date;
}

public void setDate(String date){
    this.date = date;
}

public String getDirection(){
    return direction;
}

public void setDirection(String direction){
    this.direction = direction;
}

public String getModel(){
    return model;
}

public void setModel(String model){
    this.model = model;
}

public String getAirlines(){
    return airlines;
}

public void setAirlines(String airlines){
    this.airlines = airlines;
}

public String getFrom(){
    return from;
}

public void setFrom(String from){
    this.from = from;
}

public String getTo(){
    return to;
}

public void setTo(String to){
    this.to = to;
}

}
</code>```

And finally my activity. This is how I populate listview from the array, I'd like something similar for my database.

```<code>import android.app.Activity;
import android.content.res.Resources;
import android.os.Bundle;
import android.widget.ArrayAdapter;
import android.widget.ListView;

public class DatabaseCheck extends Activity {

    private ListView lvPlanes;
    private String[] planes;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_database_check);
        lvPlanes = (ListView) findViewById(R.id.list_all);
        initResources();
        initListView();
    }

    private void initResources() {
        Resources res = getResources();
        planes = res.getStringArray(R.array.listofplanes);
    }

    private void initListView() {
        lvPlanes.setAdapter(new ArrayAdapter<String>(
                getBaseContext(),
                android.R.layout.simple_list_item_1,
                planes));
    }
}
</code>```

I know I should open DB, populate the list and close DB. I understand I should build a custom adapter, but I have no idea how to do what I want. I tried doing something, but every attempt ended in an app crash. Can anyone help me? Any suggestion will be appreciated, thank you in advance.

答:
Try this, define a CursorAdapter:

```<code>public class yourCursorAdapter extends CursorAdapter {
  public yourCursorAdapter(Context context, Cursor cursor) {
      super(context, cursor, 0);
  }

  @Override
  public View newView(Context context, Cursor cursor, ViewGroup parent) {
      return LayoutInflater.from(context).inflate(R.layout.item_todo, parent, false);
  }

  @Override
  public void bindView(View view, Context context, Cursor cursor) {

      // Find all fields to populate in inflated template
      // Example
      TextView tvDate = (TextView) view.findViewById(R.id.tvDate);

      // Extract all needed properties from cursor
      // Example
      String date = cursor.getString(cursor.getColumnIndexOrThrow("date"));

      // Populate all the required fields with extracted properties
      // Example
      tvDate.setText(date);

  }
}
</code>```

Then, in your initListView set the adapter to the listview:

```<code>private void initListView() {
     DatabaseHelper handler = new DatabaseHelper(this);
     SQLiteDatabase db = handler.getWritableDatabase();
     Cursor cursor = db.rawQuery(YOUR_SQL_QUERY, null);
     yourCursorAdapter adapter = new yourCursorAdapter(this, cursor);
     lvPlanes.setAdapter(adapter);
}
</code>```

Hope it helps

Body:
I am developing an app in Android and have a problem with passing data from databse to listview and viewing it.

Here is my database class with all methods:

package com.example.admin.planesproject;

import android.content.ContentValues;
import android.content.Context;
import android.database.Cursor;
import android.database.sqlite.SQLiteDatabase;
import android.database.sqlite.SQLiteOpenHelper;
import com.example.admin.planesproject.TodoTask;

public class PlanesDb {

private static final int DB_VERSION = 1;
private static final String DB_NAME = "database.db";
private static final String DB_TODO_TABLE = "PlanesDb";

//columns

public static final String KEY_ID = "_id";
public static final String ID_OPTIONS = "INTEGER PRIMARY KEY AUTOINCREMENT";
public static final int ID_COLUMN = 0;

public static final String KEY_DATE = "date";
public static final String DATE_OPTIONS = "TEXT NOT NULL";
public static final int DATE_COLUMN = 1;

public static final String KEY_DIRECTION = "direction";
public static final String DIRECTION_OPTIONS = "TEXT NOT NULL";
public static final int DIRECTION_COLUMN = 2;

public static final String KEY_MODEL = "model";
public static final String MODEL_OPTIONS = "TEXT";
public static final int MODEL_COLUMN = 3;

public static final String KEY_AIRLINES = "airlines";
public static final String AIRLINES_OPTIONS = "TEXT";
public static final int AIRLINES_COLUMN = 4;

public static final String KEY_FROM = "from";
public static final String FROM_OPTIONS = "TEXT";
public static final int FROM_COLUMN = 5;

public static final String KEY_TO = "to";
public static final String TO_OPTIONS = "TEXT";
public static final int TO_COLUMN = 6;

private static final String DB_CREATE_TODO_TABLE =
        "CREATE TABLE " + DB_TODO_TABLE + "( " +
                KEY_ID + " " + ID_OPTIONS + ", " +
                KEY_DATE + " " + DATE_OPTIONS + ", " +
                KEY_DIRECTION + " " + DIRECTION_OPTIONS + ", " +
                KEY_MODEL + " " + MODEL_OPTIONS + ", " +
                KEY_AIRLINES + " " + AIRLINES_OPTIONS + ", " +
                KEY_FROM + " " + FROM_OPTIONS + ", " +
                KEY_TO + " " + TO_OPTIONS + ");";

private static final String DROP_TODO_TABLE =
        "DROP TABLE IF EXISTS " + DB_TODO_TABLE;

private SQLiteDatabase db;
private Context context;
private DatabaseHelper dbHelper;

private static class DatabaseHelper extends SQLiteOpenHelper {
    public DatabaseHelper(Context context, String name,
                          SQLiteDatabase.CursorFactory factory, int version) {
        super(context, name, factory, version);
    }

    @Override
    public void onCreate(SQLiteDatabase db) {
        db.execSQL(DB_CREATE_TODO_TABLE);
    }

    @Override
    public void onUpgrade(SQLiteDatabase db, int oldVersion, int newVersion) {
        db.execSQL(DROP_TODO_TABLE);
        onCreate(db);
    }
}

public PlanesDb(Context context) {
    this.context = context;
}

public PlanesDb open(){
    dbHelper = new DatabaseHelper(context, DB_NAME, null, DB_VERSION);
    db = dbHelper.getWritableDatabase();
    return this;
}

public void close() {
    dbHelper.close();
}

public long insertTodo(String date, String direction, String model, String airlines, String from, String to) {
    ContentValues newTodoValues = new ContentValues();
    newTodoValues.put(KEY_DATE, date);
    newTodoValues.put(KEY_DIRECTION, direction);
    newTodoValues.put(KEY_MODEL, model);
    newTodoValues.put(KEY_AIRLINES, airlines);
    newTodoValues.put(KEY_FROM, from);
    newTodoValues.put(KEY_TO, to);
    return db.insert(DB_TODO_TABLE, null, newTodoValues);
}

public boolean updateTodo(TodoTask task) {
    long id = task.getId();
    String date = task.getDate();
    String direction = task.getDirection();
    String model = task.getModel();
    String airlines = task.getAirlines();
    String from = task.getFrom();
    String to = task.getTo();
    return updateTodo(id, date, direction, model, airlines, from, to);
}

public boolean updateTodo(long id, String date, String direction, String model, String airlines, String from, String to) {
    String where = KEY_ID + "=" + id;
    ContentValues updateTodoValues = new ContentValues();
    updateTodoValues.put(KEY_DATE, date);
    updateTodoValues.put(KEY_DIRECTION, direction);
    updateTodoValues.put(KEY_MODEL, model);
    updateTodoValues.put(KEY_AIRLINES, airlines);
    updateTodoValues.put(KEY_FROM, from);
    updateTodoValues.put(KEY_TO, to);
    return db.update(DB_TODO_TABLE, updateTodoValues, where, null) > 0;
}

public boolean deleteTodo(long id){
    String where = KEY_ID + "=" + id;
    return db.delete(DB_TODO_TABLE, where, null) > 0;
}

public Cursor getAllTodos() {
    String[] columns = {KEY_ID, KEY_DATE, KEY_DIRECTION, KEY_MODEL, KEY_AIRLINES, KEY_FROM, KEY_TO};
    return db.query(DB_TODO_TABLE, columns, null, null, null, null, null);
}
}

I'd like to have a listview showing everything from my table, but I cannot grasp how to do it. Tutorials don't help much as most of them only shows how to work with one row database.

My ToDo Task with all getters and setters:

public class TodoTask {
private long id;
private String date;
private String direction;
private String model;
private String airlines;
private String from;
private String to;

public TodoTask(long id, String date, String direction, String model, String airlines, String from, String to) {
    this.id = id;
    this.date = date;
    this.direction = direction;
    this.model = model;
    this.airlines = airlines;
    this.from = from;
    this.to = to;
}

public long getId() {
    return id;
}

public void setId(long id) {
    this.id = id;
}

public String getDate(){
    return date;
}

public void setDate(String date){
    this.date = date;
}

public String getDirection(){
    return direction;
}

public void setDirection(String direction){
    this.direction = direction;
}

public String getModel(){
    return model;
}

public void setModel(String model){
    this.model = model;
}

public String getAirlines(){
    return airlines;
}

public void setAirlines(String airlines){
    this.airlines = airlines;
}

public String getFrom(){
    return from;
}

public void setFrom(String from){
    this.from = from;
}

public String getTo(){
    return to;
}

public void setTo(String to){
    this.to = to;
}

}

And finally my activity. This is how I populate listview from the array, I'd like something similar for my database.

import android.app.Activity;
import android.content.res.Resources;
import android.os.Bundle;
import android.widget.ArrayAdapter;
import android.widget.ListView;

public class DatabaseCheck extends Activity {

    private ListView lvPlanes;
    private String[] planes;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_database_check);
        lvPlanes = (ListView) findViewById(R.id.list_all);
        initResources();
        initListView();
    }

    private void initResources() {
        Resources res = getResources();
        planes = res.getStringArray(R.array.listofplanes);
    }

    private void initListView() {
        lvPlanes.setAdapter(new ArrayAdapter<String>(
                getBaseContext(),
                android.R.layout.simple_list_item_1,
                planes));
    }
}

I know I should open DB, populate the list and close DB. I understand I should build a custom adapter, but I have no idea how to do what I want. I tried doing something, but every attempt ended in an app crash. Can anyone help me? Any suggestion will be appreciated, thank you in advance.

Body:
Try this, define a CursorAdapter:

public class yourCursorAdapter extends CursorAdapter {
  public yourCursorAdapter(Context context, Cursor cursor) {
      super(context, cursor, 0);
  }

  @Override
  public View newView(Context context, Cursor cursor, ViewGroup parent) {
      return LayoutInflater.from(context).inflate(R.layout.item_todo, parent, false);
  }

  @Override
  public void bindView(View view, Context context, Cursor cursor) {

      // Find all fields to populate in inflated template
      // Example
      TextView tvDate = (TextView) view.findViewById(R.id.tvDate);

      // Extract all needed properties from cursor
      // Example
      String date = cursor.getString(cursor.getColumnIndexOrThrow("date"));

      // Populate all the required fields with extracted properties
      // Example
      tvDate.setText(date);

  }
}

Then, in your initListView set the adapter to the listview:

private void initListView() {
     DatabaseHelper handler = new DatabaseHelper(this);
     SQLiteDatabase db = handler.getWritableDatabase();
     Cursor cursor = db.rawQuery(YOUR_SQL_QUERY, null);
     yourCursorAdapter adapter = new yourCursorAdapter(this, cursor);
     lvPlanes.setAdapter(adapter);
}

Hope it helps


问:
Unrecognized selector error on swipe left gesture swift
I am attempting to create a simple swift left segue to the next view controller, however I am getting an unrecognized selector error.  This the code failing:

```<code>   override func viewDidLoad() {
        super.viewDidLoad()

        //below creates the instances for swiping to change screens
        //only added right here because it's the main screen

          var swipeLeft : UISwipeGestureRecognizer = UISwipeGestureRecognizer(target: self, action: Selector("swipe:"))
        swipeLeft.direction = UISwipeGestureRecognizerDirection.Left

        self.view.addGestureRecognizer(swipeLeft)

        func swipe(Sender: UISwipeGestureRecognizer!) {
            print("swiped left")
            let vc = self.storyboard!.instantiateViewControllerWithIdentifier("OnDeck") as! OnDeck
            self.presentViewController(vc, animated: true, completion: nil)
        }
}
</code>```

This is the error I am getting:

```<code>2017-06-07 19:07:00.990 Test[65722:3911717] -[Test.GameView swipe:]: unrecognized selector sent to instance 0x7fc2d310f700
2017-06-07 19:07:01.000 Test[65722:3911717] *** Terminating app due to uncaught exception 'NSInvalidArgumentException', reason: '-[Test.GameView swipe:]: unrecognized selector sent to instance 0x7fc2d310f700'
</code>```

答:
In your case for Swift 3 the selector should look like this:

```<code>#selector(swipe(Sender:))
</code>```

You can use the #selector notation to get compile-time checks.

For Swift 2 the selector would be:

```<code>#selector(swipe(_:))
</code>```

Body:
I am attempting to create a simple swift left segue to the next view controller, however I am getting an unrecognized selector error.  This the code failing:

   override func viewDidLoad() {
        super.viewDidLoad()

        //below creates the instances for swiping to change screens
        //only added right here because it's the main screen

          var swipeLeft : UISwipeGestureRecognizer = UISwipeGestureRecognizer(target: self, action: Selector("swipe:"))
        swipeLeft.direction = UISwipeGestureRecognizerDirection.Left

        self.view.addGestureRecognizer(swipeLeft)

        func swipe(Sender: UISwipeGestureRecognizer!) {
            print("swiped left")
            let vc = self.storyboard!.instantiateViewControllerWithIdentifier("OnDeck") as! OnDeck
            self.presentViewController(vc, animated: true, completion: nil)
        }
}

This is the error I am getting:

2017-06-07 19:07:00.990 Test[65722:3911717] -[Test.GameView swipe:]: unrecognized selector sent to instance 0x7fc2d310f700
2017-06-07 19:07:01.000 Test[65722:3911717] *** Terminating app due to uncaught exception 'NSInvalidArgumentException', reason: '-[Test.GameView swipe:]: unrecognized selector sent to instance 0x7fc2d310f700'

Body:
In your case for Swift 3 the selector should look like this:

#selector(swipe(Sender:))

You can use the #selector notation to get compile-time checks.

For Swift 2 the selector would be:

#selector(swipe(_:))


问:
Unit testing event driven javascript
I'm writing a server-side application with node.js and would like to include unit tests. One thing I'm struggling with is a good way to unit test EventEmitter and setInterval/setTimout

What options do I have to unit test asynchronous server side javascript?

I'm aware I can just attach another listener to the EventEmitter that is the testing function but then how do I garantuee that the testing function ever runs? part of the unit test is to ensure every part of the unit test runs.

I could use setTimeout myself but that seems like a hackish solution.

If it helps here is some exampler code I'm trying to test.

```<code>...
function init(param) {
    ...
    // run update entities periodically
    setInterval(this._updateEntities.bind(this, param.containerFull),
        1000 / param.fps);
    ...
}
...
EntityUpdater.prototype = {
    ...
    "_updateEntities": function updateEntitiesfunc(fn) {
        this._tickEmitter.emit(
            "tick",
            new dataContainer.DataContainer(this.getEntityCount())
            .on(
                "full", fn.bind(this)
            )
        );
    },
    ...
}
...
</code>```

(emit will trigger a event)

[Edit]

I started reading some of the EvevntEmitter tests at https://github.com/ry/node/tree/master/test/simple and it helps me see how to go about this.

答:
Personally what helped me the most was reading the tests for node.js themselves (I only read about half of them).

That gave me a good feeling of how to test the asynchronous code.

Thanks to @tmdean for pointing out another example of how to test async code.

ContentLicense: CC BY-SA 2.5

Title: Unit testing event driven javascript

Body:
I'm writing a server-side application with node.js and would like to include unit tests. One thing I'm struggling with is a good way to unit test EventEmitter and setInterval/setTimout

What options do I have to unit test asynchronous server side javascript?

I'm aware I can just attach another listener to the EventEmitter that is the testing function but then how do I garantuee that the testing function ever runs? part of the unit test is to ensure every part of the unit test runs.

I could use setTimeout myself but that seems like a hackish solution.

If it helps here is some exampler code I'm trying to test.

...
function init(param) {
    ...
    // run update entities periodically
    setInterval(this._updateEntities.bind(this, param.containerFull),
        1000 / param.fps);
    ...
}
...
EntityUpdater.prototype = {
    ...
    "_updateEntities": function updateEntitiesfunc(fn) {
        this._tickEmitter.emit(
            "tick",
            new dataContainer.DataContainer(this.getEntityCount())
            .on(
                "full", fn.bind(this)
            )
        );
    },
    ...
}
...

(emit will trigger a event)

[Edit]

I started reading some of the EvevntEmitter tests at https://github.com/ry/node/tree/master/test/simple and it helps me see how to go about this.


问:
NullPointerException on set label text
Whenever I try to set the text of a label from a different ViewController on a storyboard, i get a NullPointerException:
System.NullReferenceException: Object reference not set to an instance of an object in the line UIApplication.Main(args, null, "AppDelegate");

Both of the view controllers are using the same class.
This is my code:

```<code>   async partial void LoginBtn_TouchUpInside(UIButton sender)
   {
       loadingIndicator.StartAnimating();
       var uname = username.Text;
       var pwd = password.Text;
       Uri loginUri = new Uri($"http://<loginapi>:8080/api/login?user={uname}&pwd={pwd}");
       WebClient c = new WebClient();
       string[] userinfo = parseJsonToArray(await c.DownloadStringTaskAsync(loginUri));
       welcomeLabel.Text = "Welcome, " + userinfo[0];  //index 0 of userinfo contains the user's full name.
       loadingIndicator.StopAnimating();
       PerformSegue("loggedInSegue", this);
   }
</code>```

I have verified that it is getting the correct information by logging it to the console, and everything else works perfectly fine if I comment out the welcomeLabel.Text part.

答:
You can enable all exception in your Visual Studio or Xamarin Studio so that you can break at line contain the null reference. Check this to enable all exceptions
http://www.stefandevo.com/2015/12/20/xamarin-studio-tip-break-on-all-exceptions/</p>

Title: NullPointerException on set label text

Body:
Whenever I try to set the text of a label from a different ViewController on a storyboard, i get a NullPointerException:
System.NullReferenceException: Object reference not set to an instance of an object in the line UIApplication.Main(args, null, "AppDelegate");

Both of the view controllers are using the same class.
This is my code:

   async partial void LoginBtn_TouchUpInside(UIButton sender)
   {
       loadingIndicator.StartAnimating();
       var uname = username.Text;
       var pwd = password.Text;
       Uri loginUri = new Uri($"http://<loginapi>:8080/api/login?user={uname}&pwd={pwd}");
       WebClient c = new WebClient();
       string[] userinfo = parseJsonToArray(await c.DownloadStringTaskAsync(loginUri));
       welcomeLabel.Text = "Welcome, " + userinfo[0];  //index 0 of userinfo contains the user's full name.
       loadingIndicator.StopAnimating();
       PerformSegue("loggedInSegue", this);
   }

I have verified that it is getting the correct information by logging it to the console, and everything else works perfectly fine if I comment out the welcomeLabel.Text part.