#vulnerable code 
                              @Override
  @PreFetch(enabled = false)
  public void onMessage(String instanceName, List<Message> messages,
      NotificationContext changeContext) {
    logger.info("START: GenericClusterController.onMessage() for cluster " + _clusterName);
    if (changeContext == null || changeContext.getType() != Type.CALLBACK) {
      _cache.requireFullRefresh();
    } else {
      _cache.updateDataChange(ChangeType.MESSAGE);
    }

    ClusterEvent event = new ClusterEvent(_clusterName, ClusterEventType.MessageChange);
    event.addAttribute(AttributeName.helixmanager.name(), changeContext.getManager());
    event.addAttribute(AttributeName.instanceName.name(), instanceName);
    event.addAttribute(AttributeName.changeContext.name(), changeContext);

    _eventQueue.put(event);

    if (_clusterStatusMonitor != null && messages != null) {
      _clusterStatusMonitor.addMessageQueueSize(instanceName, messages.size());
    }

    logger.info("END: GenericClusterController.onMessage() for cluster " + _clusterName);
  } 
                            #location 20 
                            #vulnerability type THREAD_SAFETY_VIOLATION
#fixed code
                              @Override
  @PreFetch(enabled = false)
  public void onMessage(String instanceName, List<Message> messages,
      NotificationContext changeContext) {
    logger.info("START: GenericClusterController.onMessage() for cluster " + _clusterName);
    if (changeContext == null || changeContext.getType() != Type.CALLBACK) {
      _cache.requireFullRefresh();
      _taskCache.requireFullRefresh();
    } else {
      _cache.updateDataChange(ChangeType.MESSAGE);
      _taskCache.updateDataChange(ChangeType.MESSAGE);
    }

    ClusterEvent event = new ClusterEvent(_clusterName, ClusterEventType.MessageChange);
    event.addAttribute(AttributeName.helixmanager.name(), changeContext.getManager());
    event.addAttribute(AttributeName.instanceName.name(), instanceName);
    event.addAttribute(AttributeName.changeContext.name(), changeContext);

    _eventQueue.put(event);
    _taskEventQueue.put(event.clone());

    if (_clusterStatusMonitor != null && messages != null) {
      _clusterStatusMonitor.addMessageQueueSize(instanceName, messages.size());
    }

    logger.info("END: GenericClusterController.onMessage() for cluster " + _clusterName);
  }