public <T> Collector<T> createSegmentBasedCollector(long id, Map<Address, Collection<Integer>> backups, int topologyId) {
      if (backups.isEmpty()) {
         return new PrimaryOwnerOnlyCollector<>();
      }
      SegmentBasedCollector<T> collector = new SegmentBasedCollector<>(id, backups, topologyId);
      BaseAckTarget prev = collectorMap.put(id, collector);
      //is it possible the have a previous collector when the topology changes after the first collector is created
      //in that case, the previous collector must have a lower topology id
      assert prev == null || prev.topologyId < topologyId : format("replaced old collector '%s' by '%s'", prev, collector);
      if (trace) {
         log.tracef("Created new collector for %s. BackupSegments=%s", id, backups);
      }
      return collector;
   }
public < T > Collector < T > createSegmentBasedCollector ( long id , Map < Address , Collection < Integer > > backups , int topologyId ) { if ( backups . isEmpty ( ) ) { return new PrimaryOwnerOnlyCollector < > ( ) ; } SegmentBasedCollector < T > collector = new SegmentBasedCollector < > ( id , backups , topologyId ) ; BaseAckTarget prev = collectorMap . put ( id , collector ) ; assert prev == null || prev . topologyId < topologyId : format ( "replaced old collector '%s' by '%s'" , prev , collector ) ; if ( trace ) { log . tracef ( "Created new collector for %s. BackupSegments=%s" , id , backups ) ; } return collector ; }
Creates a collector for {@link org.infinispan.commands.write.PutMapCommand}.

@param id         the id from {@link CommandInvocationId#getId()}.
@param backups    a map between a backup owner and its segments affected.
@param topologyId the current topology id.