### Question:
CarreraConsumer extends BaseCarreraConsumerPool { public CarreraConsumer(CarreraConfig config) { super(config); } CarreraConsumer(CarreraConfig config); synchronized void startConsume(MessageProcessor processor, int concurrency); synchronized void startConsume(MessageProcessor processor, int concurrency, Map<String, Integer> extraConcurrency); synchronized void startConsume(BatchMessageProcessor processor, int concurrency); synchronized void startConsume(BatchMessageProcessor processor, int concurrency, Map<String, Integer> extraConcurrency); @Deprecated List<ConsumeStats> getConsumeStats(); @Deprecated List<ConsumeStats> getConsumeStats(String topic); @Deprecated void close(); @Override synchronized void stop(); static final String CARRERA_CONSUMER_TYPE; }### Answer:
@Test public void testCarreraConsumer() throws InterruptedException { CountDownLatch latch = new CountDownLatch(1); carreraConsumer.startConsume(new BatchMessageProcessor() { @Override public Map<Result, List<Long>> process(List<Message> messages, Context context) { System.out.println("receive msg:" + context.toString() + ", size=" + messages.size()); Map<Result, List<Long>> map = new HashMap<>(); List<Long> list = new ArrayList<>(); for(Message msg : messages) { list.add(msg.getOffset()); } map.put(Result.SUCCESS, list); return map; } }, 10); latch.await(); }