@Test public void testCollectFail() { long start = System.currentTimeMillis(); int timeout = 10; int error = 100; List<ConsumerMessage<String>> msgs = holder.collect(1, timeout).getMessages(); assertTrue(System.currentTimeMillis() - start < config.getScanIntervalMax() + timeout + error); assertEquals(0, msgs.size()); }
@Override public PulledBatch<T> collect(int maxMessageCount, int timeout) { long startTime = System.currentTimeMillis(); Transaction t = Cat.newTransaction(CatConstants.TYPE_MESSAGE_CONSUME_COLLECT_TRIED, m_topic + ":" + m_group); try { PulledBatch<T> batch = retrive(maxMessageCount, timeout, RetrivePolicy.FAVOUR_MORE_MESSAGE); if (batch.getMessages() != null && !batch.getMessages().isEmpty()) { CatUtil.logElapse(CatConstants.TYPE_MESSAGE_CONSUME_COLLECT_ELAPSE, m_topic + ":" + m_group, startTime, batch.getMessages().size(), null, Transaction.SUCCESS); } return batch; } finally { t.setStatus(Transaction.SUCCESS); t.complete(); } }
DefaultPullConsumerHolder implements PullConsumerHolder<T>, MessageListener<T> { @Override public PulledBatch<T> collect(int maxMessageCount, int timeout) { long startTime = System.currentTimeMillis(); Transaction t = Cat.newTransaction(CatConstants.TYPE_MESSAGE_CONSUME_COLLECT_TRIED, m_topic + ":" + m_group); try { PulledBatch<T> batch = retrive(maxMessageCount, timeout, RetrivePolicy.FAVOUR_MORE_MESSAGE); if (batch.getMessages() != null && !batch.getMessages().isEmpty()) { CatUtil.logElapse(CatConstants.TYPE_MESSAGE_CONSUME_COLLECT_ELAPSE, m_topic + ":" + m_group, startTime, batch.getMessages().size(), null, Transaction.SUCCESS); } return batch; } finally { t.setStatus(Transaction.SUCCESS); t.complete(); } } }
DefaultPullConsumerHolder implements PullConsumerHolder<T>, MessageListener<T> { @Override public PulledBatch<T> collect(int maxMessageCount, int timeout) { long startTime = System.currentTimeMillis(); Transaction t = Cat.newTransaction(CatConstants.TYPE_MESSAGE_CONSUME_COLLECT_TRIED, m_topic + ":" + m_group); try { PulledBatch<T> batch = retrive(maxMessageCount, timeout, RetrivePolicy.FAVOUR_MORE_MESSAGE); if (batch.getMessages() != null && !batch.getMessages().isEmpty()) { CatUtil.logElapse(CatConstants.TYPE_MESSAGE_CONSUME_COLLECT_ELAPSE, m_topic + ":" + m_group, startTime, batch.getMessages().size(), null, Transaction.SUCCESS); } return batch; } finally { t.setStatus(Transaction.SUCCESS); t.complete(); } } DefaultPullConsumerHolder(String topic, String groupId, int partitionCount, PullConsumerConfig config,
	      AckManager ackManager, OffsetStorage offsetStorage, ConsumerConfig consumerConfig); }
DefaultPullConsumerHolder implements PullConsumerHolder<T>, MessageListener<T> { @Override public PulledBatch<T> collect(int maxMessageCount, int timeout) { long startTime = System.currentTimeMillis(); Transaction t = Cat.newTransaction(CatConstants.TYPE_MESSAGE_CONSUME_COLLECT_TRIED, m_topic + ":" + m_group); try { PulledBatch<T> batch = retrive(maxMessageCount, timeout, RetrivePolicy.FAVOUR_MORE_MESSAGE); if (batch.getMessages() != null && !batch.getMessages().isEmpty()) { CatUtil.logElapse(CatConstants.TYPE_MESSAGE_CONSUME_COLLECT_ELAPSE, m_topic + ":" + m_group, startTime, batch.getMessages().size(), null, Transaction.SUCCESS); } return batch; } finally { t.setStatus(Transaction.SUCCESS); t.complete(); } } DefaultPullConsumerHolder(String topic, String groupId, int partitionCount, PullConsumerConfig config,
	      AckManager ackManager, OffsetStorage offsetStorage, ConsumerConfig consumerConfig); @Override void onMessage(List<ConsumerMessage<T>> msgs); @Override PulledBatch<T> poll(int maxMessageCount, int timeout); @Override PulledBatch<T> collect(int maxMessageCount, int timeout); @Override void close(); void setConsumerHolder(ConsumerHolder consumerHolder); }
DefaultPullConsumerHolder implements PullConsumerHolder<T>, MessageListener<T> { @Override public PulledBatch<T> collect(int maxMessageCount, int timeout) { long startTime = System.currentTimeMillis(); Transaction t = Cat.newTransaction(CatConstants.TYPE_MESSAGE_CONSUME_COLLECT_TRIED, m_topic + ":" + m_group); try { PulledBatch<T> batch = retrive(maxMessageCount, timeout, RetrivePolicy.FAVOUR_MORE_MESSAGE); if (batch.getMessages() != null && !batch.getMessages().isEmpty()) { CatUtil.logElapse(CatConstants.TYPE_MESSAGE_CONSUME_COLLECT_ELAPSE, m_topic + ":" + m_group, startTime, batch.getMessages().size(), null, Transaction.SUCCESS); } return batch; } finally { t.setStatus(Transaction.SUCCESS); t.complete(); } } DefaultPullConsumerHolder(String topic, String groupId, int partitionCount, PullConsumerConfig config,
	      AckManager ackManager, OffsetStorage offsetStorage, ConsumerConfig consumerConfig); @Override void onMessage(List<ConsumerMessage<T>> msgs); @Override PulledBatch<T> poll(int maxMessageCount, int timeout); @Override PulledBatch<T> collect(int maxMessageCount, int timeout); @Override void close(); void setConsumerHolder(ConsumerHolder consumerHolder);  }