ContentBasedBlockingQueueRecordWriter implements RecordWriter<P> { @Override public void writeRecords(Batch<P> batch) throws Exception { DefaultPredicate<P> defaultPredicate = new DefaultPredicate<>(); BlockingQueue<Record<P>> defaultQueue = queueMap.get(defaultPredicate); for (Record<P> record : batch) { boolean matched = false; for (Map.Entry<Predicate<P>, BlockingQueue<Record<P>>> entry : queueMap.entrySet()) { Predicate<P> predicate = entry.getKey(); BlockingQueue<Record<P>> queue = entry.getValue(); if (!(predicate instanceof DefaultPredicate) && predicate.matches(record)) { queue.put(record); matched = true; break; } } if (!matched && defaultQueue != null) { defaultQueue.put(record); } } } ContentBasedBlockingQueueRecordWriter(Map<Predicate<P>, BlockingQueue<Record<P>>> queueMap); @Override void writeRecords(Batch<P> batch); Map<Predicate<P>, BlockingQueue<Record<P>>> getQueueMap();  }
@Test public void orangeRecordShouldBeWrittenToOrangeQueue() throws Exception { recordWriter.writeRecords(new Batch(orangeRecord, appleRecord)); assertThat(orangeQueue).containsOnly(orangeRecord); assertThat(defaultQueue).containsOnly(appleRecord); }