This method uses synchronization because it can be used during indexing,
and Druid can call aggregate() and get() concurrently
https://github.com/apache/incubator-druid/pull/3956
168113
@Override
  public void aggregate()
  {
    final IndexedInts keys = keySelector.getRow();
    for (int i = 0; i < valueSelectors.length; i++) {
      values[i] = valueSelectors[i].getDouble();
    }
    synchronized (this) {
      for (int i = 0, keysSize = keys.size(); i < keysSize; i++) {
        final String key = keySelector.lookupName(keys.get(i));
        sketch.update(key, values);
      }
    }
  }
60041
public function getAggregate(string $uuid, string $aggregateClass, int $user): AggregateInterface
    {
        return $this->aggregateFactory->build($uuid, $aggregateClass, null, $user);
    }
26548
public boolean acquireProcessLock() throws MongobeeConnectionException, MongobeeLockException {
    verifyDbConnection();
    boolean acquired = lockDao.acquireLock(getMongoDatabase());

    if (!acquired && waitForLock) {
      long timeToGiveUp = new Date().getTime() + (changeLogLockWaitTime * 1000 * 60);
      while (!acquired && new Date().getTime() < timeToGiveUp) {
        acquired = lockDao.acquireLock(getMongoDatabase());
        if (!acquired) {
          logger.info("Waiting for changelog lock....");
          try {
            Thread.sleep(changeLogLockPollRate * 1000);
          } catch (InterruptedException e) {
            // nothing
          }
        }
      }
    }

    if (!acquired && throwExceptionIfCannotObtainLock) {
      logger.info("Mongobee did not acquire process lock. Throwing exception.");
      throw new MongobeeLockException("Could not acquire process lock");
    }

    return acquired;
  }
960420
function(token,options){
      options = options||{};
      options.token = token;
      var obj = {
        type:"rest"
        ,args:{ 
          url: '/v1/sync',
          data: options,
          method: 'get' 
        }
      };

      var s = through();

      repipe(s,function(err,last,done){
        o.log('repipe got error? ',err,' should i repipe?');
        if(err && err.code != 'E_MDM') return done(err);

        getConnection(function(err,con){
          if(err) return done(err);
          var dbstream = con.mdm.createReadStream(obj);

          var noendstream = dbstream.pipe(through(function(data){
            this.queue(data);
          },function(){
            if(options.tail || options.tail == undefined) {
              // repipe resumes the stream on error events!

              var error = new Error("never end bro!");
              error.code = "E_MDM";

              this.emit("error",error);
            } else {
              this.queue(null);
            }
          }));

          dbstream.on('error',function(err){
            noendstream.emit("error",err)
          });

          done(false,noendstream);
        });
      });

      s.on('data',function(d){
        // on reconnect start sync from where i left off.
        obj.args.data.start = d.time;
      })
      return s;
    }
662142
function multiread(stream, get) {
    if (stream._multiread === undefined) {
        stream._multiread = stream._multiread || [];
        stream._multireading = stream._multireading || 0;
        stream._multireadmax = stream._multireadmax || concurrency;
    }
    if (stream._multiread.length) {
        push();
    } else {
        stream.once('_push', push);
    }
    while ((stream._multiread.length + stream._multireading) < stream._multireadmax) {
        stream._multireading++;
        if (!get(done)) break;
    }

    function done(item) {
        stream._multireading--;
        if (item === null && !stream._multireading) {
            stream.emit('_push');
        } else if (item) {
            stream._multiread.push(item);
            stream.emit('_push');
        }
    }

    function push() {
        if (stream._multiread.length) {
            stream.push(stream._multiread.shift());
        } else {
            stream.push(null);
        }
    }
}
923425
function(syncIt,dataset,indexes,options) {
		this._syncIt = syncIt;
		this._dataset = dataset;
		this._jreadCache = [];
		this._lastOptions = undefined;
		this._hideDeleted = true;
		this._index = {
			id:{},
			s:{},
			k:{},
			v:{},
			m:{},
			r:{},
			t:{}
		};
		for (var i=0;i<indexes.length;i++) {
			this._index[indexes[i]] = {};
		}
		if ((typeof options == 'object') && (options.hasOwnProperty('hideDeleted'))) {
			this._hideDeleted = options.hideDeleted ? true : false;
		}
	}
386447
function runSubqueries (subQueries, callback, collection) {
        var i,
            responsesReceived = 0,
            length = subQueries.length,
            joinedSet = [];//The array where the results are going to get stuffed

        if (subQueries.length > 0) {
            for (i = 0; i < subQueries.length; i += 1) {

                collection.aggregate(subQueries[i], function (err, results) {
                    joinedSet = joinedSet.concat(results);
                    responsesReceived += 1;

                    if (responsesReceived === length) {
                        callback(joinedSet);
                    }
                });
            }
        } else {
            callback([]);
        }

        return joinedSet;
    }
855210
@Override
  public void aggregate(final ByteBuffer buf, final int position)
  {
    final ArrayOfDoublesSketch update = selector.getObject();
    if (update == null) {
      return;
    }
    // Wrapping memory and ArrayOfDoublesUnion is inexpensive compared to union operations.
    // Maintaining a cache of wrapped objects per buffer position like in Theta sketch aggregator
    // might might be considered, but it would increase complexity including relocate() support.
    final WritableMemory mem = WritableMemory.wrap(buf, ByteOrder.LITTLE_ENDIAN);
    final WritableMemory region = mem.writableRegion(position, maxIntermediateSize);
    final Lock lock = stripedLock.getAt(ArrayOfDoublesSketchBuildBufferAggregator.lockIndex(position)).writeLock();
    lock.lock();
    try {
      final ArrayOfDoublesUnion union = ArrayOfDoublesSketches.wrapUnion(region);
      union.update(update);
    }
    finally {
      lock.unlock();
    }
  }
1073346
async def import_aggregated(self, async_client, pid):
        
        self._logger.info('Importing: {}'.format(pid))

        task_set = set()

        object_info_pyxb = d1_common.types.dataoneTypes.ObjectInfo()
        object_info_pyxb.identifier = pid
        task_set.add(self.import_object(async_client, object_info_pyxb))

        result_set, task_set = await asyncio.wait(task_set)

        assert len(result_set) == 1
        assert not task_set

        sysmeta_pyxb = result_set.pop().result()

        if not sysmeta_pyxb:
            # Import was skipped
            return

        assert d1_common.xml.get_req_val(sysmeta_pyxb.identifier) == pid

        if d1_gmn.app.did.is_resource_map_db(pid):
            for member_pid in d1_gmn.app.resource_map.get_resource_map_members_by_map(
                pid
            ):
                self.progress_logger.event("Importing aggregated SciObj")
                self._logger.info('Importing aggregated SciObj. pid="{}"'.format(pid))
                await self.import_aggregated(async_client, member_pid)
714775
private JsonParserIterator<DataSegment> getMetadataSegments(
      DruidLeaderClient coordinatorClient,
      ObjectMapper jsonMapper,
      BytesAccumulatingResponseHandler responseHandler,
      Set<String> watchedDataSources
  )
  {
    String query = "/druid/coordinator/v1/metadata/segments";
    if (watchedDataSources != null && !watchedDataSources.isEmpty()) {
      log.debug(
          "filtering datasources in published segments based on broker's watchedDataSources[%s]", watchedDataSources);
      final StringBuilder sb = new StringBuilder();
      for (String ds : watchedDataSources) {
        sb.append("datasources=").append(ds).append("&");
      }
      sb.setLength(sb.length() - 1);
      query = "/druid/coordinator/v1/metadata/segments?" + sb;
    }
    Request request;
    try {
      request = coordinatorClient.makeRequest(
          HttpMethod.GET,
          StringUtils.format(query),
          false
      );
    }
    catch (IOException e) {
      throw new RuntimeException(e);
    }
    ListenableFuture<InputStream> future = coordinatorClient.goAsync(
        request,
        responseHandler
    );

    final JavaType typeRef = jsonMapper.getTypeFactory().constructType(new TypeReference<DataSegment>()
    {
    });
    return new JsonParserIterator<>(
        typeRef,
        future,
        request.getUrl().toString(),
        null,
        request.getUrl().getHost(),
        jsonMapper,
        responseHandler
    );
  }
1141491
private Void indexUpdate(Collection<Consumer<Void>> state, Consumer<Void> indexAction) {
        if (async) {
            state.add(indexAction);
        } else {
            indexAction.accept(null);
        }
        return null;
    }
1326633
function(cacheNum) {
      if (cacheNum === undefined) {
        cacheNum = 1;
      }
      if (!uuidCache.length) {
        ajax({url: this.urlPrefix + "/_uuids", data: {count: cacheNum}, async:
              false}, {
            success: function(resp) {
              uuidCache = resp.uuids;
            }
          },
          "Failed to retrieve UUID batch."
        );
      }
      return uuidCache.shift();
    }
1349846
@Override
  public void aggregate(final ByteBuffer buf, final int position)
  {
    for (int i = 0; i < valueSelectors.length; i++) {
      values[i] = valueSelectors[i].getDouble();
    }
    final IndexedInts keys = keySelector.getRow();
    // Wrapping memory and ArrayOfDoublesSketch is inexpensive compared to sketch operations.
    // Maintaining a cache of wrapped objects per buffer position like in Theta sketch aggregator
    // might might be considered, but it would increase complexity including relocate() support.
    final WritableMemory mem = WritableMemory.wrap(buf, ByteOrder.LITTLE_ENDIAN);
    final WritableMemory region = mem.writableRegion(position, maxIntermediateSize);
    final Lock lock = stripedLock.getAt(lockIndex(position)).writeLock();
    lock.lock();
    try {
      final ArrayOfDoublesUpdatableSketch sketch = ArrayOfDoublesSketches.wrapUpdatableSketch(region);
      for (int i = 0, keysSize = keys.size(); i < keysSize; i++) {
        final String key = keySelector.lookupName(keys.get(i));
        sketch.update(key, values);
      }
    }
    finally {
      lock.unlock();
    }
  }
106470
function AggregateByKey(sc, dependencies, reducer, combiner, init, args) {
  Dataset.call(this, sc, dependencies);
  this.combiner = combiner;
  this.reducer = reducer;
  this.init = init;
  this.args = args;
  this.shuffling = true;
  this.executed = false;
  this.buffer = {};
  this.type = 'AggregateByKey';
}
1005078
function(cursor, callback) {
    groupedDelta = {};

    async.series([
      (cb) => {
        logger.info('Fetching delta', {category: 'sync-remote-delta'});

        fetchDelta({cursor}, (err, newCursor) => {
          if(err) return cb(err);

          cb(null, newCursor);
        });
      },
      (cb) => {
        logger.info('Applying collections which need to be redownloaded', {category: 'sync-remote-delta'});

        syncDb.find({$and: [{directory: true}, {downloadOriginal: true}]}, (err, syncedNodes) => {
          if(err) return cb(err);
          if(!syncedNodes) return cb(null);

          async.map(syncedNodes, (syncedNode, mapCb) => {
            if(!syncedNode.remoteId) mapCb(null);

            fetchDelta({id: syncedNode.remoteId}, mapCb);
          }, cb);
        });
      }
    ], (err, results) => {
      logger.info('getDelta ended', {category: 'sync-remote-delta'});

      callback(err, results[0]);
    });
  }
523720
protected void retrieveAggregateRoot(final Set<String> removedNodeIds, final Map<String, NodeData> map)
   {
      if (indexingConfig != null)
      {
         AggregateRule[] aggregateRules = indexingConfig.getAggregateRules();
         if (aggregateRules == null)
         {
            return;
         }
         long time = 0;
         if (log.isDebugEnabled())
         {
            time = System.currentTimeMillis();
         }
         int found = SecurityHelper.doPrivilegedAction(new PrivilegedAction<Integer>()
         {
            public Integer run()
            {
               int found = 0;
               try
               {
                  CachingMultiIndexReader reader = indexRegister.getDefaultIndex().getIndexReader();
                  try
                  {
                     Term aggregateUUIDs = new Term(FieldNames.AGGREGATED_NODE_UUID, "");
                     TermDocs tDocs = reader.termDocs();
                     try
                     {
                        ItemDataConsumer ism = getContext().getItemStateManager();
                        for (Iterator<String> it = removedNodeIds.iterator(); it.hasNext();)
                        {
                           String id = it.next();
                           aggregateUUIDs = aggregateUUIDs.createTerm(id);
                           tDocs.seek(aggregateUUIDs);
                           while (tDocs.next())
                           {
                              Document doc = reader.document(tDocs.doc(), FieldSelectors.UUID);
                              String uuid = doc.get(FieldNames.UUID);
                              ItemData itd = ism.getItemData(uuid);
                              if (itd == null)
                              {
                                 continue;
                              }
                              if (!itd.isNode())
                              {
                                 throw new RepositoryException("Item with id:" + uuid + " is not a node");
                              }
                              map.put(uuid, (NodeData)itd);
                              found++;
                           }
                        }
                     }
                     finally
                     {
                        tDocs.close();
                     }
                  }
                  finally
                  {
                     reader.release();
                  }
               }
               catch (Exception e)
               {
                  log.warn("Exception while retrieving aggregate roots", e);
               }
               return found;
            }
         });
         if (log.isDebugEnabled())
         {
            time = System.currentTimeMillis() - time;
            log.debug("Retrieved {} aggregate roots in {} ms.", new Integer(found), new Long(time));
         }
      }
   }
1071877
@Override
    public R visitSynchronized(SynchronizedTree node, P p) {
        R r = scan(node.getExpression(), p);
        r = scanAndReduce(node.getBlock(), p, r);
        return r;
    }