#fixed code
                             private void run() {

		
		Properties props = new Properties();
		props.put("zookeeper.hosts", "10.252.5.131");
		props.put("zookeeper.port", "2181");
		props.put("zookeeper.broker.path", "/brokers");
		props.put("kafka.topic", "valid_subpub");
		props.put("kafka.consumer.id", "valid_subpub");		
		props.put("zookeeper.consumer.connection", "10.252.5.113:2182");
		props.put("zookeeper.consumer.path", "/kafka-new");
		
		SparkConf _sparkConf = new SparkConf().setAppName("KafkaReceiver")
				.set("spark.streaming.receiver.writeAheadLog.enable", "false");;

		JavaStreamingContext ssc = new JavaStreamingContext(_sparkConf,
				new Duration(10000));
		
		//Specify number of Receivers you need. 
		//It should be less than or equal to number of Partitions of your topic
		
		int numberOfReceivers = 2;

		JavaDStream<MessageAndMetadata> unionStreams = ReceiverLauncher.launch(ssc, props, numberOfReceivers);

		unionStreams
				.foreachRDD(new Function2<JavaRDD<MessageAndMetadata>, Time, Void>() {

					@Override
					public Void call(JavaRDD<MessageAndMetadata> rdd,
							Time time) throws Exception {

						System.out.println("Number of records in this Batch is " + rdd.count());
						return null;
					}
				});
		
		ssc.start();
		ssc.awaitTermination();
	}
#vulnerable code 
                             private void run() {

		String checkpointDirectory = "hdfs://10.252.5.113:9000/user/hadoop/spark";
		int _partitionCount = 3;

		List<JavaDStream<MessageAndMetadata>> streamsList = new ArrayList<JavaDStream<MessageAndMetadata>>(
				_partitionCount);
		JavaDStream<MessageAndMetadata> unionStreams;

		SparkConf _sparkConf = new SparkConf().setAppName("KafkaReceiver")
				.set("spark.streaming.receiver.writeAheadLog.enable", "true");;

		JavaStreamingContext ssc = new JavaStreamingContext(_sparkConf,
				new Duration(10000));

		for (int i = 0; i < _partitionCount; i++) {

			streamsList.add(ssc.receiverStream(new KafkaReceiver(_props, i)));

		}

		// Union all the streams if there is more than 1 stream
		if (streamsList.size() > 1) {
			unionStreams = ssc.union(streamsList.get(0),
					streamsList.subList(1, streamsList.size()));
		} else {
			// Otherwise, just use the 1 stream
			unionStreams = streamsList.get(0);
		}

		unionStreams.checkpoint(new Duration(10000));

		try {
			unionStreams
					.foreachRDD(new Function2<JavaRDD<MessageAndMetadata>, Time, Void>() {

						@Override
						public Void call(JavaRDD<MessageAndMetadata> rdd,
								Time time) throws Exception {

							for (MessageAndMetadata record : rdd.collect()) {

								if (record != null) {

									System.out.println(new String(record
											.getPayload()));

								}

							}
							return null;
						}
					});
		} catch (Exception ex) {

			ex.printStackTrace();
		}

		ssc.checkpoint(checkpointDirectory);
		ssc.start();
		ssc.awaitTermination();
	} 
                            #location 61 
                            #vulnerability type RESOURCE_LEAK