#vulnerable code 
                              void enqueueData(ZNRecordUpdate e)
  {
    if(!_initialized || _shutdownFlag)
    {
      LOG.error("inited:" + _initialized + " shutdownFlag:"+_shutdownFlag+" , return");
      return;
    }
    // Do local merge if receive multiple update on the same path
    synchronized(_dataBufferRef)
    {
      if(_dataBufferRef.get().containsKey(e.getPath()))
      {
        ZNRecord oldVal = _dataBufferRef.get().get(e.getPath()).getRecord();
        oldVal = e.getZNRecordUpdater().update(oldVal);
        _dataBufferRef.get().get(e.getPath())._record = oldVal;
      }
      else
      {
        _dataBufferRef.get().put(e.getPath(), e);
      }
    }
    if(_dataBufferRef.get().size() > MAX_UPDATE_LIMIT)
    {
      sendData();
    }
  } 
                            #location 24 
                            #vulnerability type THREAD_SAFETY_VIOLATION
#fixed code
                              void sendData()
  {
    ConcurrentHashMap<String, ZNRecordUpdate> updateCache  = null;
    
    synchronized(_dataBufferRef)
    {
      updateCache = _dataBufferRef.getAndSet(new ConcurrentHashMap<String, ZNRecordUpdate>());
    }
    
    if(updateCache != null)
    {
      List<String> paths = new ArrayList<String>();
      List<DataUpdater<ZNRecord>> updaters = new ArrayList<DataUpdater<ZNRecord>>();
      List<ZNRecord> vals = new ArrayList<ZNRecord>();
      for(ZNRecordUpdate holder : updateCache.values())
      {
        paths.add(holder.getPath());
        updaters.add(holder.getZNRecordUpdater());
        vals.add(holder.getRecord());
      }
      // Batch write the accumulated updates into zookeeper
      if(paths.size() > 0)
      {
        _accessor.updateChildren(paths, updaters, BaseDataAccessor.Option.PERSISTENT);
      }
      LOG.info("Updating " + vals.size() + " records");
    }
    else
    {
      LOG.warn("null _dataQueueRef. Should be in the beginning only");
    }
  }