code: 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;
     }
docs: 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.