#vulnerable code 
                              @Test()
  public void TestSchedulerZeroMsg() throws Exception
  {
    TestMessagingHandlerFactory factory = new TestMessagingHandlerFactory();
    HelixManager manager = null;
    for (int i = 0; i < NODE_NR; i++)
    {
      String hostDest = "localhost_" + (START_PORT + i);
      _startCMResultMap.get(hostDest)._manager.getMessagingService()
          .registerMessageHandlerFactory(factory.getMessageType(), factory);
      manager = _startCMResultMap.get(hostDest)._manager;
    }

    Message schedulerMessage = new Message(MessageType.SCHEDULER_MSG + "", UUID
        .randomUUID().toString());
    schedulerMessage.setTgtSessionId("*");
    schedulerMessage.setTgtName("CONTROLLER");
    // TODO: change it to "ADMIN" ?
    schedulerMessage.setSrcName("CONTROLLER");

    // Template for the individual message sent to each participant
    Message msg = new Message(factory.getMessageType(), "Template");
    msg.setTgtSessionId("*");
    msg.setMsgState(MessageState.NEW);

    // Criteria to send individual messages
    Criteria cr = new Criteria();
    cr.setInstanceName("localhost_DOESNOTEXIST");
    cr.setRecipientInstanceType(InstanceType.PARTICIPANT);
    cr.setSessionSpecific(false);
    cr.setResource("%");
    cr.setPartition("%");

    ObjectMapper mapper = new ObjectMapper();
    SerializationConfig serializationConfig = mapper.getSerializationConfig();
    serializationConfig.set(SerializationConfig.Feature.INDENT_OUTPUT, true);

    StringWriter sw = new StringWriter();
    mapper.writeValue(sw, cr);

    String crString = sw.toString();

    schedulerMessage.getRecord().setSimpleField("Criteria", crString);
    schedulerMessage.getRecord().setMapField("MessageTemplate",
        msg.getRecord().getSimpleFields());
    schedulerMessage.getRecord().setSimpleField("TIMEOUT", "-1");

    HelixDataAccessor helixDataAccessor = manager.getHelixDataAccessor();
    Builder keyBuilder = helixDataAccessor.keyBuilder();
    PropertyKey controllerMessageKey = keyBuilder
        .controllerMessage(schedulerMessage.getMsgId());
    helixDataAccessor.setProperty(controllerMessageKey, schedulerMessage);

    Thread.sleep(3000);

    Assert.assertEquals(0, factory._results.size());
    PropertyKey controllerTaskStatus = keyBuilder.controllerTaskStatus(
        MessageType.SCHEDULER_MSG.toString(), schedulerMessage.getMsgId());
    for(int i = 0; i< 10; i++)
    {
      if(helixDataAccessor.getProperty(controllerTaskStatus) == null)
      {
        Thread.sleep(1000);
      }
    }
    ZNRecord statusUpdate = helixDataAccessor.getProperty(controllerTaskStatus)
        .getRecord();
    Assert.assertTrue(statusUpdate.getMapField("SentMessageCount")
        .get("MessageCount").equals("0"));
    int count = 0;
    for (Set<String> val : factory._results.values())
    {
      count += val.size();
    }
    Assert.assertEquals(count, 0);
  } 
                            #location 67 
                            #vulnerability type NULL_DEREFERENCE
#fixed code
                              @Test()
  public void TestSchedulerZeroMsg() throws Exception
  {
    TestMessagingHandlerFactory factory = new TestMessagingHandlerFactory();
    HelixManager manager = null;
    for (int i = 0; i < NODE_NR; i++)
    {
      String hostDest = "localhost_" + (START_PORT + i);
      _startCMResultMap.get(hostDest)._manager.getMessagingService()
          .registerMessageHandlerFactory(factory.getMessageType(), factory);
      manager = _startCMResultMap.get(hostDest)._manager;
    }

    Message schedulerMessage = new Message(MessageType.SCHEDULER_MSG + "", UUID
        .randomUUID().toString());
    schedulerMessage.setTgtSessionId("*");
    schedulerMessage.setTgtName("CONTROLLER");
    // TODO: change it to "ADMIN" ?
    schedulerMessage.setSrcName("CONTROLLER");

    // Template for the individual message sent to each participant
    Message msg = new Message(factory.getMessageType(), "Template");
    msg.setTgtSessionId("*");
    msg.setMsgState(MessageState.NEW);

    // Criteria to send individual messages
    Criteria cr = new Criteria();
    cr.setInstanceName("localhost_DOESNOTEXIST");
    cr.setRecipientInstanceType(InstanceType.PARTICIPANT);
    cr.setSessionSpecific(false);
    cr.setResource("%");
    cr.setPartition("%");

    ObjectMapper mapper = new ObjectMapper();
    SerializationConfig serializationConfig = mapper.getSerializationConfig();
    serializationConfig.set(SerializationConfig.Feature.INDENT_OUTPUT, true);

    StringWriter sw = new StringWriter();
    mapper.writeValue(sw, cr);

    String crString = sw.toString();

    schedulerMessage.getRecord().setSimpleField("Criteria", crString);
    schedulerMessage.getRecord().setMapField("MessageTemplate",
        msg.getRecord().getSimpleFields());
    schedulerMessage.getRecord().setSimpleField("TIMEOUT", "-1");

    HelixDataAccessor helixDataAccessor = manager.getHelixDataAccessor();
    Builder keyBuilder = helixDataAccessor.keyBuilder();
    PropertyKey controllerMessageKey = keyBuilder
        .controllerMessage(schedulerMessage.getMsgId());
    helixDataAccessor.setProperty(controllerMessageKey, schedulerMessage);

    Thread.sleep(3000);

    Assert.assertEquals(0, factory._results.size());
    PropertyKey controllerTaskStatus = keyBuilder.controllerTaskStatus(
        MessageType.SCHEDULER_MSG.toString(), schedulerMessage.getMsgId());
    for(int i = 0; i< 10; i++)
    {
      StatusUpdate update = helixDataAccessor.getProperty(controllerTaskStatus);
      if(update == null || update.getRecord().getMapField("SentMessageCount") == null)
      {
        Thread.sleep(1000);
      }
    }
    ZNRecord statusUpdate = helixDataAccessor.getProperty(controllerTaskStatus)
        .getRecord();
    Assert.assertTrue(statusUpdate.getMapField("SentMessageCount")
        .get("MessageCount").equals("0"));
    int count = 0;
    for (Set<String> val : factory._results.values())
    {
      count += val.size();
    }
    Assert.assertEquals(count, 0);
  }