MySQLMessageQueueStorage implements MessageQueueStorage { @Override public void nack(Tpp tpp, String groupId, boolean resend, List<Pair<Long, Integer>> msgSeqs) { if (CollectionUtil.isNotEmpty(msgSeqs)) { Topic topic = m_metaService.findTopic(tpp.getTopic()); String retryPolicyValue = topic.findConsumerGroup(groupId).getRetryPolicy(); if (retryPolicyValue == null || "".equals(retryPolicyValue.trim())) { retryPolicyValue = topic.getConsumerRetryPolicy(); } RetryPolicy retryPolicy = RetryPolicyFactory.create(retryPolicyValue); List<Pair<Long, Integer>> toDeadLetter = new ArrayList<>(); List<Pair<Long, Integer>> toResend = new ArrayList<>(); for (Pair<Long, Integer> pair : msgSeqs) { if (resend) { pair.setValue(pair.getValue() - 1); } else { pair.setValue(retryPolicy.getRetryTimes()); } if (pair.getValue() <= 0) { toDeadLetter.add(pair); } else { toResend.add(pair); } } try { copyToDeadLetter(tpp, groupId, toDeadLetter, resend); copyToResend(tpp, groupId, toResend, resend, retryPolicy); } catch (DalException e) { e.printStackTrace(); } } }  @Override void appendMessages(Tpp tpp, Collection<MessageRawDataBatch> batches); @Override synchronized Object findLastOffset(Tpp tpp, int groupId); @Override FetchResult fetchMessages(Tpp tpp, Object startOffset, int batchSize); @Override void nack(Tpp tpp, String groupId, boolean resend, List<Pair<Long, Integer>> msgSeqs); @Override void ack(Tpp tpp, String groupId, boolean resend, long msgSeq); @Override synchronized Object findLastResendOffset(Tpg tpg); @SuppressWarnings("unchecked") @Override FetchResult fetchResendMessages(Tpg tpg, Object startOffset, int batchSize);  }
@Test public void testNackMessage() throws Exception { Tpp tpp = new Tpp("order_new", 0, true); s.nack(tpp, "group1", false, Arrays.asList(new Pair<>(1L, 1), new Pair<>(2L, 2))); }
@Test public void testNackResend() throws Exception { Tpp tpp = new Tpp("order_new", 0, true); s.nack(tpp, "group1", true, Arrays.asList(new Pair<>(1L, 1), new Pair<>(2L, 2))); }