class SummaryCollector implements UpsertResultCollector {

        private final Map<String, String> nodeInfo;

        private final Object lock = new Object();

        SummaryCollector(Map<String, String> nodeInfo) {
            this.nodeInfo = nodeInfo;
        }

        @Override
        public Supplier<UpsertResults> supplier() {
            return () -> new UpsertResults(nodeInfo);
        }

        @Override
        public Accumulator accumulator() {
            return this::processShardResponse;
        }

        @Override
        public BinaryOperator<UpsertResults> combiner() {
            return (i, o) -> {
                synchronized (lock) {
                    i.merge(o);
                }
                return i;
            };
        }

        @Override
        public Function<UpsertResults, Iterable<Row>> finisher() {
            return UpsertResults::rowsIterable;
        }

        void processShardResponse(UpsertResults upsertResults,
                                  ShardResponse shardResponse,
                                  List<RowSourceInfo> rowSourceInfos) {<FILL_FUNCTION_BODY>}
    }

            synchronized (lock) {
                List<ShardResponse.Failure> failures = shardResponse.failures();
                IntArrayList locations = shardResponse.itemIndices();
                for (int i = 0; i < failures.size(); i++) {
                    ShardResponse.Failure failure = failures.get(i);
                    int location = locations.get(i);
                    RowSourceInfo rowSourceInfo = rowSourceInfos.get(location);
                    String msg = failure == null ? null : failure.message();
                    upsertResults.addResult(rowSourceInfo.sourceUri, msg, rowSourceInfo.lineNumber);
                }
            }
        
<no_super_class>