MySQLMessageQueueStorage implements MessageQueueStorage { @Override public void appendMessages(Tpp tpp, Collection<MessageRawDataBatch> batches) throws Exception { List<MessagePriority> msgs = new ArrayList<>(); for (MessageRawDataBatch batch : batches) { List<PartialDecodedMessage> pdmsgs = batch.getMessages(); for (PartialDecodedMessage pdmsg : pdmsgs) { MessagePriority msg = new MessagePriority(); msg.setAttributes(pdmsg.readDurableProperties()); msg.setCreationDate(new Date(pdmsg.getBornTime())); msg.setPartition(tpp.getPartition()); msg.setPayload(pdmsg.readBody()); msg.setPriority(tpp.isPriority() ? 0 : 1); msg.setProducerId(101); msg.setProducerIp("1.1.1.1"); msg.setRefKey(pdmsg.getKey()); msg.setTopic(tpp.getTopic()); msg.setCodecType(pdmsg.getBodyCodecType()); msgs.add(msg); } } m_msgDao.insert(msgs.toArray(new MessagePriority[msgs.size()])); }  @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 testAppendMessages() throws Exception { String topic = "order_new"; Tpp tpp = new Tpp(topic, 0, true); Collection<MessageRawDataBatch> batches = new ArrayList<>(); ByteBuf rawData = Unpooled.buffer(); MessageRawDataBatch batch = new MessageRawDataBatch(topic, Arrays.asList(1), rawData); batches.add(batch); s.appendMessages(tpp, batches); }