@Test public void emptyChannel() throws Exception { mongoSink.process(); assertThat(fongo.getDB("test").getCollection("test").count()).isEqualTo(0); }
@Override public Status process() throws EventDeliveryException { log.debug("Executing MongoSink.process()..."); Status status = Status.READY; Channel channel = getChannel(); Transaction txn = channel.getTransaction(); try { txn.begin(); int count; List<Event> eventList= new ArrayList<Event>(); for (count = 0; count < batchSize; ++count) { Event event = channel.take(); if (event == null) { break; } eventList.add(event); } if (count <= 0) { sinkCounter.incrementBatchEmptyCount(); counterGroup.incrementAndGet("channel.underflow"); status = Status.BACKOFF; } else { if (count < batchSize) { sinkCounter.incrementBatchUnderflowCount(); status = Status.BACKOFF; } else { sinkCounter.incrementBatchCompleteCount(); } for (Event event : eventList) { final DBObject document = this.eventParser.parse(event); if (this.updateInsteadReplace && document.get("_id") != null) { BasicDBObject searchQuery = new BasicDBObject().append("_id", document.get("_id")); BasicDBObject updatedDocument = new BasicDBObject().append("$set", document); getDBCollection(event).update(searchQuery,updatedDocument,true,false); } else { getDBCollection(event).save(document); } } sinkCounter.addToEventDrainAttemptCount(eventList.size()); } txn.commit(); sinkCounter.addToEventDrainSuccessCount(count); counterGroup.incrementAndGet("transaction.success"); } catch (ChannelException e) { log.error("Unexpected error while executing MongoSink.process", e); txn.rollback(); status = Status.BACKOFF; this.sinkCounter.incrementConnectionFailedCount(); } catch (Throwable t) { log.error("Unexpected error while executing MongoSink.process", t); txn.rollback(); status = Status.BACKOFF; if (t instanceof Error) { throw new MongoSinkException(t); } } finally { txn.close(); } return status; }
MongoSink extends AbstractSink implements Configurable { @Override public Status process() throws EventDeliveryException { log.debug("Executing MongoSink.process()..."); Status status = Status.READY; Channel channel = getChannel(); Transaction txn = channel.getTransaction(); try { txn.begin(); int count; List<Event> eventList= new ArrayList<Event>(); for (count = 0; count < batchSize; ++count) { Event event = channel.take(); if (event == null) { break; } eventList.add(event); } if (count <= 0) { sinkCounter.incrementBatchEmptyCount(); counterGroup.incrementAndGet("channel.underflow"); status = Status.BACKOFF; } else { if (count < batchSize) { sinkCounter.incrementBatchUnderflowCount(); status = Status.BACKOFF; } else { sinkCounter.incrementBatchCompleteCount(); } for (Event event : eventList) { final DBObject document = this.eventParser.parse(event); if (this.updateInsteadReplace && document.get("_id") != null) { BasicDBObject searchQuery = new BasicDBObject().append("_id", document.get("_id")); BasicDBObject updatedDocument = new BasicDBObject().append("$set", document); getDBCollection(event).update(searchQuery,updatedDocument,true,false); } else { getDBCollection(event).save(document); } } sinkCounter.addToEventDrainAttemptCount(eventList.size()); } txn.commit(); sinkCounter.addToEventDrainSuccessCount(count); counterGroup.incrementAndGet("transaction.success"); } catch (ChannelException e) { log.error("Unexpected error while executing MongoSink.process", e); txn.rollback(); status = Status.BACKOFF; this.sinkCounter.incrementConnectionFailedCount(); } catch (Throwable t) { log.error("Unexpected error while executing MongoSink.process", t); txn.rollback(); status = Status.BACKOFF; if (t instanceof Error) { throw new MongoSinkException(t); } } finally { txn.close(); } return status; } }
MongoSink extends AbstractSink implements Configurable { @Override public Status process() throws EventDeliveryException { log.debug("Executing MongoSink.process()..."); Status status = Status.READY; Channel channel = getChannel(); Transaction txn = channel.getTransaction(); try { txn.begin(); int count; List<Event> eventList= new ArrayList<Event>(); for (count = 0; count < batchSize; ++count) { Event event = channel.take(); if (event == null) { break; } eventList.add(event); } if (count <= 0) { sinkCounter.incrementBatchEmptyCount(); counterGroup.incrementAndGet("channel.underflow"); status = Status.BACKOFF; } else { if (count < batchSize) { sinkCounter.incrementBatchUnderflowCount(); status = Status.BACKOFF; } else { sinkCounter.incrementBatchCompleteCount(); } for (Event event : eventList) { final DBObject document = this.eventParser.parse(event); if (this.updateInsteadReplace && document.get("_id") != null) { BasicDBObject searchQuery = new BasicDBObject().append("_id", document.get("_id")); BasicDBObject updatedDocument = new BasicDBObject().append("$set", document); getDBCollection(event).update(searchQuery,updatedDocument,true,false); } else { getDBCollection(event).save(document); } } sinkCounter.addToEventDrainAttemptCount(eventList.size()); } txn.commit(); sinkCounter.addToEventDrainSuccessCount(count); counterGroup.incrementAndGet("transaction.success"); } catch (ChannelException e) { log.error("Unexpected error while executing MongoSink.process", e); txn.rollback(); status = Status.BACKOFF; this.sinkCounter.incrementConnectionFailedCount(); } catch (Throwable t) { log.error("Unexpected error while executing MongoSink.process", t); txn.rollback(); status = Status.BACKOFF; if (t instanceof Error) { throw new MongoSinkException(t); } } finally { txn.close(); } return status; } MongoSink(); }
MongoSink extends AbstractSink implements Configurable { @Override public Status process() throws EventDeliveryException { log.debug("Executing MongoSink.process()..."); Status status = Status.READY; Channel channel = getChannel(); Transaction txn = channel.getTransaction(); try { txn.begin(); int count; List<Event> eventList= new ArrayList<Event>(); for (count = 0; count < batchSize; ++count) { Event event = channel.take(); if (event == null) { break; } eventList.add(event); } if (count <= 0) { sinkCounter.incrementBatchEmptyCount(); counterGroup.incrementAndGet("channel.underflow"); status = Status.BACKOFF; } else { if (count < batchSize) { sinkCounter.incrementBatchUnderflowCount(); status = Status.BACKOFF; } else { sinkCounter.incrementBatchCompleteCount(); } for (Event event : eventList) { final DBObject document = this.eventParser.parse(event); if (this.updateInsteadReplace && document.get("_id") != null) { BasicDBObject searchQuery = new BasicDBObject().append("_id", document.get("_id")); BasicDBObject updatedDocument = new BasicDBObject().append("$set", document); getDBCollection(event).update(searchQuery,updatedDocument,true,false); } else { getDBCollection(event).save(document); } } sinkCounter.addToEventDrainAttemptCount(eventList.size()); } txn.commit(); sinkCounter.addToEventDrainSuccessCount(count); counterGroup.incrementAndGet("transaction.success"); } catch (ChannelException e) { log.error("Unexpected error while executing MongoSink.process", e); txn.rollback(); status = Status.BACKOFF; this.sinkCounter.incrementConnectionFailedCount(); } catch (Throwable t) { log.error("Unexpected error while executing MongoSink.process", t); txn.rollback(); status = Status.BACKOFF; if (t instanceof Error) { throw new MongoSinkException(t); } } finally { txn.close(); } return status; } MongoSink(); @Override void configure(Context context); @Override Status process(); @Override synchronized void start(); @Override synchronized void stop(); }
MongoSink extends AbstractSink implements Configurable { @Override public Status process() throws EventDeliveryException { log.debug("Executing MongoSink.process()..."); Status status = Status.READY; Channel channel = getChannel(); Transaction txn = channel.getTransaction(); try { txn.begin(); int count; List<Event> eventList= new ArrayList<Event>(); for (count = 0; count < batchSize; ++count) { Event event = channel.take(); if (event == null) { break; } eventList.add(event); } if (count <= 0) { sinkCounter.incrementBatchEmptyCount(); counterGroup.incrementAndGet("channel.underflow"); status = Status.BACKOFF; } else { if (count < batchSize) { sinkCounter.incrementBatchUnderflowCount(); status = Status.BACKOFF; } else { sinkCounter.incrementBatchCompleteCount(); } for (Event event : eventList) { final DBObject document = this.eventParser.parse(event); if (this.updateInsteadReplace && document.get("_id") != null) { BasicDBObject searchQuery = new BasicDBObject().append("_id", document.get("_id")); BasicDBObject updatedDocument = new BasicDBObject().append("$set", document); getDBCollection(event).update(searchQuery,updatedDocument,true,false); } else { getDBCollection(event).save(document); } } sinkCounter.addToEventDrainAttemptCount(eventList.size()); } txn.commit(); sinkCounter.addToEventDrainSuccessCount(count); counterGroup.incrementAndGet("transaction.success"); } catch (ChannelException e) { log.error("Unexpected error while executing MongoSink.process", e); txn.rollback(); status = Status.BACKOFF; this.sinkCounter.incrementConnectionFailedCount(); } catch (Throwable t) { log.error("Unexpected error while executing MongoSink.process", t); txn.rollback(); status = Status.BACKOFF; if (t instanceof Error) { throw new MongoSinkException(t); } } finally { txn.close(); } return status; } MongoSink(); @Override void configure(Context context); @Override Status process(); @Override synchronized void start(); @Override synchronized void stop();  }