@Test public void testEmpty() { @SuppressWarnings({"unchecked", "rawtypes"}) StreamMerger merger = new StreamMerger(Arrays.asList(new Integer(0)), 0); Assert.assertNull(merger.getNextTuple()); }
public T getNextTuple() { long minTsFound = Long.MAX_VALUE; boolean eachBufferFilled = true; Integer minTsPartitionNumber = null; Iterator<Entry<Integer, LinkedList<T>>> it = this.mergeBuffer.entrySet().iterator(); while(it.hasNext()) { Entry<Integer, LinkedList<T>> partition = it.next(); LinkedList<T> partitionBuffer = partition.getValue(); try { long ts = this.getTsValue(partitionBuffer.getFirst()); assert (ts >= this.latestTs); if(ts == this.latestTs) { logger.trace("Extract tuple with same timestamp (partition, tuple): {}, {}", partition.getKey(), partitionBuffer.getFirst()); return partitionBuffer.removeFirst(); } if(ts < minTsFound) { minTsFound = ts; minTsPartitionNumber = partition.getKey(); } } catch(NoSuchElementException e) { if(this.disabledPartitions.contains(partition.getKey())) { logger.trace("Closing empty and disabled parition: {}", partition.getKey()); it.remove(); } else { logger.trace("Found empty parition: {}", partition.getKey()); eachBufferFilled = false; } } } if(eachBufferFilled && minTsPartitionNumber != null) { logger.trace("Extract tuple min timestamp (ts, partition, tuple): {}, {}, {}", new Long(minTsFound), minTsPartitionNumber, this.mergeBuffer.get(minTsPartitionNumber).getFirst()); this.latestTs = minTsFound; return this.mergeBuffer.get(minTsPartitionNumber).removeFirst(); } logger.trace("Could not extract tuple."); return null; }
StreamMerger { public T getNextTuple() { long minTsFound = Long.MAX_VALUE; boolean eachBufferFilled = true; Integer minTsPartitionNumber = null; Iterator<Entry<Integer, LinkedList<T>>> it = this.mergeBuffer.entrySet().iterator(); while(it.hasNext()) { Entry<Integer, LinkedList<T>> partition = it.next(); LinkedList<T> partitionBuffer = partition.getValue(); try { long ts = this.getTsValue(partitionBuffer.getFirst()); assert (ts >= this.latestTs); if(ts == this.latestTs) { logger.trace("Extract tuple with same timestamp (partition, tuple): {}, {}", partition.getKey(), partitionBuffer.getFirst()); return partitionBuffer.removeFirst(); } if(ts < minTsFound) { minTsFound = ts; minTsPartitionNumber = partition.getKey(); } } catch(NoSuchElementException e) { if(this.disabledPartitions.contains(partition.getKey())) { logger.trace("Closing empty and disabled parition: {}", partition.getKey()); it.remove(); } else { logger.trace("Found empty parition: {}", partition.getKey()); eachBufferFilled = false; } } } if(eachBufferFilled && minTsPartitionNumber != null) { logger.trace("Extract tuple min timestamp (ts, partition, tuple): {}, {}, {}", new Long(minTsFound), minTsPartitionNumber, this.mergeBuffer.get(minTsPartitionNumber).getFirst()); this.latestTs = minTsFound; return this.mergeBuffer.get(minTsPartitionNumber).removeFirst(); } logger.trace("Could not extract tuple."); return null; } }
StreamMerger { public T getNextTuple() { long minTsFound = Long.MAX_VALUE; boolean eachBufferFilled = true; Integer minTsPartitionNumber = null; Iterator<Entry<Integer, LinkedList<T>>> it = this.mergeBuffer.entrySet().iterator(); while(it.hasNext()) { Entry<Integer, LinkedList<T>> partition = it.next(); LinkedList<T> partitionBuffer = partition.getValue(); try { long ts = this.getTsValue(partitionBuffer.getFirst()); assert (ts >= this.latestTs); if(ts == this.latestTs) { logger.trace("Extract tuple with same timestamp (partition, tuple): {}, {}", partition.getKey(), partitionBuffer.getFirst()); return partitionBuffer.removeFirst(); } if(ts < minTsFound) { minTsFound = ts; minTsPartitionNumber = partition.getKey(); } } catch(NoSuchElementException e) { if(this.disabledPartitions.contains(partition.getKey())) { logger.trace("Closing empty and disabled parition: {}", partition.getKey()); it.remove(); } else { logger.trace("Found empty parition: {}", partition.getKey()); eachBufferFilled = false; } } } if(eachBufferFilled && minTsPartitionNumber != null) { logger.trace("Extract tuple min timestamp (ts, partition, tuple): {}, {}, {}", new Long(minTsFound), minTsPartitionNumber, this.mergeBuffer.get(minTsPartitionNumber).getFirst()); this.latestTs = minTsFound; return this.mergeBuffer.get(minTsPartitionNumber).removeFirst(); } logger.trace("Could not extract tuple."); return null; } StreamMerger(Collection<Integer> partitionIds, int tsIndex);  StreamMerger(Collection<Integer> partitionIds, String tsAttributeName);  StreamMerger(Collection<Integer> partitionIds, TimeStampExtractor<T> tsExtractor); }
StreamMerger { public T getNextTuple() { long minTsFound = Long.MAX_VALUE; boolean eachBufferFilled = true; Integer minTsPartitionNumber = null; Iterator<Entry<Integer, LinkedList<T>>> it = this.mergeBuffer.entrySet().iterator(); while(it.hasNext()) { Entry<Integer, LinkedList<T>> partition = it.next(); LinkedList<T> partitionBuffer = partition.getValue(); try { long ts = this.getTsValue(partitionBuffer.getFirst()); assert (ts >= this.latestTs); if(ts == this.latestTs) { logger.trace("Extract tuple with same timestamp (partition, tuple): {}, {}", partition.getKey(), partitionBuffer.getFirst()); return partitionBuffer.removeFirst(); } if(ts < minTsFound) { minTsFound = ts; minTsPartitionNumber = partition.getKey(); } } catch(NoSuchElementException e) { if(this.disabledPartitions.contains(partition.getKey())) { logger.trace("Closing empty and disabled parition: {}", partition.getKey()); it.remove(); } else { logger.trace("Found empty parition: {}", partition.getKey()); eachBufferFilled = false; } } } if(eachBufferFilled && minTsPartitionNumber != null) { logger.trace("Extract tuple min timestamp (ts, partition, tuple): {}, {}, {}", new Long(minTsFound), minTsPartitionNumber, this.mergeBuffer.get(minTsPartitionNumber).getFirst()); this.latestTs = minTsFound; return this.mergeBuffer.get(minTsPartitionNumber).removeFirst(); } logger.trace("Could not extract tuple."); return null; } StreamMerger(Collection<Integer> partitionIds, int tsIndex);  StreamMerger(Collection<Integer> partitionIds, String tsAttributeName);  StreamMerger(Collection<Integer> partitionIds, TimeStampExtractor<T> tsExtractor); void addTuple(Integer partitionNumber, T t); T getNextTuple(); boolean closePartition(Integer partitionId); int getNumberOpenPartitions(); void disablePartition(Integer partitionId); }
StreamMerger { public T getNextTuple() { long minTsFound = Long.MAX_VALUE; boolean eachBufferFilled = true; Integer minTsPartitionNumber = null; Iterator<Entry<Integer, LinkedList<T>>> it = this.mergeBuffer.entrySet().iterator(); while(it.hasNext()) { Entry<Integer, LinkedList<T>> partition = it.next(); LinkedList<T> partitionBuffer = partition.getValue(); try { long ts = this.getTsValue(partitionBuffer.getFirst()); assert (ts >= this.latestTs); if(ts == this.latestTs) { logger.trace("Extract tuple with same timestamp (partition, tuple): {}, {}", partition.getKey(), partitionBuffer.getFirst()); return partitionBuffer.removeFirst(); } if(ts < minTsFound) { minTsFound = ts; minTsPartitionNumber = partition.getKey(); } } catch(NoSuchElementException e) { if(this.disabledPartitions.contains(partition.getKey())) { logger.trace("Closing empty and disabled parition: {}", partition.getKey()); it.remove(); } else { logger.trace("Found empty parition: {}", partition.getKey()); eachBufferFilled = false; } } } if(eachBufferFilled && minTsPartitionNumber != null) { logger.trace("Extract tuple min timestamp (ts, partition, tuple): {}, {}, {}", new Long(minTsFound), minTsPartitionNumber, this.mergeBuffer.get(minTsPartitionNumber).getFirst()); this.latestTs = minTsFound; return this.mergeBuffer.get(minTsPartitionNumber).removeFirst(); } logger.trace("Could not extract tuple."); return null; } StreamMerger(Collection<Integer> partitionIds, int tsIndex);  StreamMerger(Collection<Integer> partitionIds, String tsAttributeName);  StreamMerger(Collection<Integer> partitionIds, TimeStampExtractor<T> tsExtractor); void addTuple(Integer partitionNumber, T t); T getNextTuple(); boolean closePartition(Integer partitionId); int getNumberOpenPartitions(); void disablePartition(Integer partitionId);  }