MySQLMessageQueueStorage implements MessageQueueStorage { @Override public synchronized Object findLastOffset(Tpp tpp, int groupId) throws Exception { String topic = tpp.getTopic(); int partition = tpp.getPartition(); int priority = tpp.getPriorityInt(); List<OffsetMessage> lastOffset = m_offsetMessageDao.find(topic, partition, priority, groupId, OffsetMessageEntity.READSET_FULL); if (lastOffset.isEmpty()) { List<MessagePriority> topMsg = m_msgDao.top(topic, partition, priority, MessagePriorityEntity.READSET_FULL); long startOffset = 0L; if (!topMsg.isEmpty()) { startOffset = CollectionUtil.last(topMsg).getId(); } OffsetMessage offset = new OffsetMessage(); offset.setCreationDate(new Date()); offset.setGroupId(groupId); offset.setOffset(startOffset); offset.setPartition(partition); offset.setPriority(priority); offset.setTopic(topic); m_offsetMessageDao.insert(offset); return offset.getOffset(); } else { return CollectionUtil.last(lastOffset).getOffset(); } }  @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 testFindLastOffset() throws Exception { Tpp tpp = new Tpp("order_new", 0, true); s.findLastOffset(tpp, 100); }