@Test public void commutativeWithMultipleUpdates() { Status newFile = StatusUtil.fileCreated(100), update1 = StatusUtil.ingestedUntil(100), update2 = StatusUtil.ingestedUntil(200), repl1 = StatusUtil.replicated(50), repl2 = StatusUtil.replicated(150); Status order1 = combiner.typedReduce(key, Arrays.asList(newFile, update1, repl1, update2, repl2).iterator()); Status permutation = combiner.typedReduce(key, Arrays.asList(newFile, repl1, update1, repl2, update2).iterator()); assertEquals(order1, permutation); permutation = combiner.typedReduce(key, Arrays.asList(newFile, repl1, repl2, update1, update2).iterator()); assertEquals(order1, permutation); permutation = combiner.typedReduce(key, Arrays.asList(newFile, update1, update2, repl1, repl2, update1, update2).iterator()); assertEquals(order1, permutation); }
@Override public Status typedReduce(Key key, Iterator<Status> iter) { Builder combined = null; while (iter.hasNext()) { Status status = iter.next(); if (combined == null) { if (iter.hasNext()) { combined = Status.newBuilder(); } else { if (log.isTraceEnabled()) { log.trace("Returned single value: {} {}", key.toStringNoTruncate(), ProtobufUtil.toString(status)); } return status; } } combine(combined, status); } if (log.isTraceEnabled()) { log.trace("Combined: {} {}", key.toStringNoTruncate(), ProtobufUtil.toString(combined.build())); } return combined.build(); }
StatusCombiner extends TypedValueCombiner<Status> { @Override public Status typedReduce(Key key, Iterator<Status> iter) { Builder combined = null; while (iter.hasNext()) { Status status = iter.next(); if (combined == null) { if (iter.hasNext()) { combined = Status.newBuilder(); } else { if (log.isTraceEnabled()) { log.trace("Returned single value: {} {}", key.toStringNoTruncate(), ProtobufUtil.toString(status)); } return status; } } combine(combined, status); } if (log.isTraceEnabled()) { log.trace("Combined: {} {}", key.toStringNoTruncate(), ProtobufUtil.toString(combined.build())); } return combined.build(); } }
StatusCombiner extends TypedValueCombiner<Status> { @Override public Status typedReduce(Key key, Iterator<Status> iter) { Builder combined = null; while (iter.hasNext()) { Status status = iter.next(); if (combined == null) { if (iter.hasNext()) { combined = Status.newBuilder(); } else { if (log.isTraceEnabled()) { log.trace("Returned single value: {} {}", key.toStringNoTruncate(), ProtobufUtil.toString(status)); } return status; } } combine(combined, status); } if (log.isTraceEnabled()) { log.trace("Combined: {} {}", key.toStringNoTruncate(), ProtobufUtil.toString(combined.build())); } return combined.build(); }  }
StatusCombiner extends TypedValueCombiner<Status> { @Override public Status typedReduce(Key key, Iterator<Status> iter) { Builder combined = null; while (iter.hasNext()) { Status status = iter.next(); if (combined == null) { if (iter.hasNext()) { combined = Status.newBuilder(); } else { if (log.isTraceEnabled()) { log.trace("Returned single value: {} {}", key.toStringNoTruncate(), ProtobufUtil.toString(status)); } return status; } } combine(combined, status); } if (log.isTraceEnabled()) { log.trace("Combined: {} {}", key.toStringNoTruncate(), ProtobufUtil.toString(combined.build())); } return combined.build(); }  @Override IteratorOptions describeOptions(); @Override boolean validateOptions(Map<String,String> options); @Override void init(SortedKeyValueIterator<Key,Value> source, Map<String,String> options,
      IteratorEnvironment env); @Override Status typedReduce(Key key, Iterator<Status> iter); void combine(Builder combined, Status status); }
StatusCombiner extends TypedValueCombiner<Status> { @Override public Status typedReduce(Key key, Iterator<Status> iter) { Builder combined = null; while (iter.hasNext()) { Status status = iter.next(); if (combined == null) { if (iter.hasNext()) { combined = Status.newBuilder(); } else { if (log.isTraceEnabled()) { log.trace("Returned single value: {} {}", key.toStringNoTruncate(), ProtobufUtil.toString(status)); } return status; } } combine(combined, status); } if (log.isTraceEnabled()) { log.trace("Combined: {} {}", key.toStringNoTruncate(), ProtobufUtil.toString(combined.build())); } return combined.build(); }  @Override IteratorOptions describeOptions(); @Override boolean validateOptions(Map<String,String> options); @Override void init(SortedKeyValueIterator<Key,Value> source, Map<String,String> options,
      IteratorEnvironment env); @Override Status typedReduce(Key key, Iterator<Status> iter); void combine(Builder combined, Status status);  }