MySQLMessageQueueStorage implements MessageQueueStorage { @Override public void ack(Tpp tpp, String groupId, boolean resend, long msgSeq) { try { String topic = tpp.getTopic(); int partition = tpp.getPartition(); int intGroupId = m_metaService.getGroupIdInt(groupId); if (resend) { OffsetResend proto = getOffsetResend(topic, partition, intGroupId); ResendGroupId resendRow = m_resendDao.findByPK(msgSeq, topic, partition, intGroupId, ResendGroupIdEntity.READSET_FULL); proto.setTopic(topic); proto.setPartition(partition); proto.setLastScheduleDate(resendRow.getScheduleDate()); proto.setLastId(resendRow.getId()); m_offsetResendDao.updateByPK(proto, OffsetResendEntity.UPDATESET_OFFSET); } else { OffsetMessage proto = getOffsetMessage(tpp, intGroupId); proto.setTopic(topic); proto.setPartition(partition); proto.setOffset(msgSeq); m_offsetMessageDao.updateByPK(proto, OffsetMessageEntity.UPDATESET_OFFSET); } } 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 testAckMessage() throws Exception { Tpp tpp = new Tpp("order_new", 0, true); s.ack(tpp, "group1", false, 222); }
@Test public void testAckResend() throws Exception { Tpp tpp = new Tpp("order_new", 0, true); s.ack(tpp, "group1", true, 1); }