MySQLMessageQueueStorage implements MessageQueueStorage { @Override public synchronized Object findLastResendOffset(Tpg tpg) throws Exception { int groupId = m_metaService.getGroupIdInt(tpg.getGroupId()); List<OffsetResend> tops = m_offsetResendDao.top(tpg.getTopic(), tpg.getPartition(), groupId, OffsetResendEntity.READSET_FULL); if (CollectionUtil.isNotEmpty(tops)) { OffsetResend top = CollectionUtil.first(tops); return new Pair<>(top.getLastScheduleDate(), top.getLastId()); } else { OffsetResend proto = new OffsetResend(); proto.setTopic(tpg.getTopic()); proto.setPartition(tpg.getPartition()); proto.setGroupId(m_metaService.getGroupIdInt(tpg.getGroupId())); proto.setLastScheduleDate(new Date(0)); proto.setLastId(0L); proto.setCreationDate(new Date()); m_offsetResendDao.insert(proto); return new Pair<>(proto.getLastScheduleDate(), proto.getLastId()); } }  @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 testFindLastResendOffset() throws Exception { Tpg tpg = new Tpg("order_new", 0, "group1"); s.findLastResendOffset(tpg); }