@Test public void testEmitAndCommit() throws InterruptedException { talosStormSpout.nextTuple(); Thread.sleep(commitInterval); talosStormSpout.nextTuple(); ArgumentCaptor<List> tupleCaptor = ArgumentCaptor.forClass(List.class); verify(collector, atLeastOnce()).emit(tupleCaptor.capture(), anyString()); List tuple = tupleCaptor.getValue(); assertEquals(message, tuple.get(0)); assertEquals(offset, tuple.get(1)); assertEquals(topic, tuple.get(2)); assertEquals(partition, tuple.get(3)); ArgumentCaptor<Long> offsetCaptor = ArgumentCaptor.forClass(Long.class); verify(msgCkp).checkpoint(offsetCaptor.capture()); assertEquals(offset, offsetCaptor.getValue().longValue()); }
@Override public void nextTuple() { if (shouldCommit()) { commitOffsets(); } boolean emitted = false; for (TalosStormMessageProcessor processor : msgProcessors.values()) { Optional<PartitionMessagePool> msgPoolOpt = processor.getMessagePool(); if (msgPoolOpt.isPresent()) { Optional<TalosStormMessage> msgOpt = msgPoolOpt.get().get(); if (msgOpt.isPresent()) { Iterable<List<Object>> tuples = talosConfig.talosScheme .generateTuples(msgOpt.get().topicPartition, msgOpt.get().message); for (List<Object> tuple : tuples) { collector.emit(tuple, msgOpt.get().id); } emitted = true; } } } if (!emitted) { try { LOG.debug(String.format("No data to emmit, TalosStormSpout will sleep %dms", idleMs)); Thread.sleep(idleMs); } catch (InterruptedException e) { } } }
TalosStormSpout extends BaseRichSpout { @Override public void nextTuple() { if (shouldCommit()) { commitOffsets(); } boolean emitted = false; for (TalosStormMessageProcessor processor : msgProcessors.values()) { Optional<PartitionMessagePool> msgPoolOpt = processor.getMessagePool(); if (msgPoolOpt.isPresent()) { Optional<TalosStormMessage> msgOpt = msgPoolOpt.get().get(); if (msgOpt.isPresent()) { Iterable<List<Object>> tuples = talosConfig.talosScheme .generateTuples(msgOpt.get().topicPartition, msgOpt.get().message); for (List<Object> tuple : tuples) { collector.emit(tuple, msgOpt.get().id); } emitted = true; } } } if (!emitted) { try { LOG.debug(String.format("No data to emmit, TalosStormSpout will sleep %dms", idleMs)); Thread.sleep(idleMs); } catch (InterruptedException e) { } } } }
TalosStormSpout extends BaseRichSpout { @Override public void nextTuple() { if (shouldCommit()) { commitOffsets(); } boolean emitted = false; for (TalosStormMessageProcessor processor : msgProcessors.values()) { Optional<PartitionMessagePool> msgPoolOpt = processor.getMessagePool(); if (msgPoolOpt.isPresent()) { Optional<TalosStormMessage> msgOpt = msgPoolOpt.get().get(); if (msgOpt.isPresent()) { Iterable<List<Object>> tuples = talosConfig.talosScheme .generateTuples(msgOpt.get().topicPartition, msgOpt.get().message); for (List<Object> tuple : tuples) { collector.emit(tuple, msgOpt.get().id); } emitted = true; } } } if (!emitted) { try { LOG.debug(String.format("No data to emmit, TalosStormSpout will sleep %dms", idleMs)); Thread.sleep(idleMs); } catch (InterruptedException e) { } } } TalosStormSpout(TalosStormConfig conf); }
TalosStormSpout extends BaseRichSpout { @Override public void nextTuple() { if (shouldCommit()) { commitOffsets(); } boolean emitted = false; for (TalosStormMessageProcessor processor : msgProcessors.values()) { Optional<PartitionMessagePool> msgPoolOpt = processor.getMessagePool(); if (msgPoolOpt.isPresent()) { Optional<TalosStormMessage> msgOpt = msgPoolOpt.get().get(); if (msgOpt.isPresent()) { Iterable<List<Object>> tuples = talosConfig.talosScheme .generateTuples(msgOpt.get().topicPartition, msgOpt.get().message); for (List<Object> tuple : tuples) { collector.emit(tuple, msgOpt.get().id); } emitted = true; } } } if (!emitted) { try { LOG.debug(String.format("No data to emmit, TalosStormSpout will sleep %dms", idleMs)); Thread.sleep(idleMs); } catch (InterruptedException e) { } } } TalosStormSpout(TalosStormConfig conf); @Override void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector); @Override void nextTuple(); @Override void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer); @Override void ack(Object msgId); @Override void fail(Object msgId); @Override void close(); }
TalosStormSpout extends BaseRichSpout { @Override public void nextTuple() { if (shouldCommit()) { commitOffsets(); } boolean emitted = false; for (TalosStormMessageProcessor processor : msgProcessors.values()) { Optional<PartitionMessagePool> msgPoolOpt = processor.getMessagePool(); if (msgPoolOpt.isPresent()) { Optional<TalosStormMessage> msgOpt = msgPoolOpt.get().get(); if (msgOpt.isPresent()) { Iterable<List<Object>> tuples = talosConfig.talosScheme .generateTuples(msgOpt.get().topicPartition, msgOpt.get().message); for (List<Object> tuple : tuples) { collector.emit(tuple, msgOpt.get().id); } emitted = true; } } } if (!emitted) { try { LOG.debug(String.format("No data to emmit, TalosStormSpout will sleep %dms", idleMs)); Thread.sleep(idleMs); } catch (InterruptedException e) { } } } TalosStormSpout(TalosStormConfig conf); @Override void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector); @Override void nextTuple(); @Override void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer); @Override void ack(Object msgId); @Override void fail(Object msgId); @Override void close();  }