@Test public void testAckResend() throws Exception { Tpp tpp = new Tpp("order_new", 0, true); s.ack(tpp, "group1", true, 1); }
@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.translateToIntGroupId(tpp.getTopic(), groupId); if (resend) { OffsetResend proto = getOffsetResend(topic, partition, intGroupId); proto.setTopic(topic); proto.setPartition(partition); proto.setLastScheduleDate(new Date()); proto.setLastId(msgSeq); m_ackFlusher.ackOffsetResend(proto); } else { OffsetMessage proto = getOffsetMessage(tpp, intGroupId); proto.setTopic(topic); proto.setPartition(partition); proto.setOffset(msgSeq); m_ackFlusher.ackOffsetMessage(proto); } } catch (DalException e) { log.error("Failed to ack messages(topic={}, partition={}, priority={}, groupId={}).", tpp.getTopic(), tpp.getPartition(), tpp.isPriority(), groupId, e); } }
MySQLMessageQueueStorage implements MessageQueueStorage, Initializable { @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.translateToIntGroupId(tpp.getTopic(), groupId); if (resend) { OffsetResend proto = getOffsetResend(topic, partition, intGroupId); proto.setTopic(topic); proto.setPartition(partition); proto.setLastScheduleDate(new Date()); proto.setLastId(msgSeq); m_ackFlusher.ackOffsetResend(proto); } else { OffsetMessage proto = getOffsetMessage(tpp, intGroupId); proto.setTopic(topic); proto.setPartition(partition); proto.setOffset(msgSeq); m_ackFlusher.ackOffsetMessage(proto); } } catch (DalException e) { log.error("Failed to ack messages(topic={}, partition={}, priority={}, groupId={}).", tpp.getTopic(), tpp.getPartition(), tpp.isPriority(), groupId, e); } } }
MySQLMessageQueueStorage implements MessageQueueStorage, Initializable { @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.translateToIntGroupId(tpp.getTopic(), groupId); if (resend) { OffsetResend proto = getOffsetResend(topic, partition, intGroupId); proto.setTopic(topic); proto.setPartition(partition); proto.setLastScheduleDate(new Date()); proto.setLastId(msgSeq); m_ackFlusher.ackOffsetResend(proto); } else { OffsetMessage proto = getOffsetMessage(tpp, intGroupId); proto.setTopic(topic); proto.setPartition(partition); proto.setOffset(msgSeq); m_ackFlusher.ackOffsetMessage(proto); } } catch (DalException e) { log.error("Failed to ack messages(topic={}, partition={}, priority={}, groupId={}).", tpp.getTopic(), tpp.getPartition(), tpp.isPriority(), groupId, e); } }  }
MySQLMessageQueueStorage implements MessageQueueStorage, Initializable { @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.translateToIntGroupId(tpp.getTopic(), groupId); if (resend) { OffsetResend proto = getOffsetResend(topic, partition, intGroupId); proto.setTopic(topic); proto.setPartition(partition); proto.setLastScheduleDate(new Date()); proto.setLastId(msgSeq); m_ackFlusher.ackOffsetResend(proto); } else { OffsetMessage proto = getOffsetMessage(tpp, intGroupId); proto.setTopic(topic); proto.setPartition(partition); proto.setOffset(msgSeq); m_ackFlusher.ackOffsetMessage(proto); } } catch (DalException e) { log.error("Failed to ack messages(topic={}, partition={}, priority={}, groupId={}).", tpp.getTopic(), tpp.getPartition(), tpp.isPriority(), groupId, e); } }  @Override void appendMessages(String topicName, int partition, boolean priority,
	      Collection<MessageBatchWithRawData> batches); @Override synchronized Object findLastOffset(Tpp tpp, int groupId); @Override FetchResult fetchMessages(Tpp tpp, List<Object> offsets); @Override FetchResult fetchMessages(Tpp tpp, Object startOffset, int batchSize, String filter); @Override void nack(Tpp tpp, String groupId, boolean resend, List<Pair<Long, MessageMeta>> msgId2Metas); @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); @Override Object findMessageOffsetByTime(Tpp tpp, long time); @Override void initialize(); }
MySQLMessageQueueStorage implements MessageQueueStorage, Initializable { @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.translateToIntGroupId(tpp.getTopic(), groupId); if (resend) { OffsetResend proto = getOffsetResend(topic, partition, intGroupId); proto.setTopic(topic); proto.setPartition(partition); proto.setLastScheduleDate(new Date()); proto.setLastId(msgSeq); m_ackFlusher.ackOffsetResend(proto); } else { OffsetMessage proto = getOffsetMessage(tpp, intGroupId); proto.setTopic(topic); proto.setPartition(partition); proto.setOffset(msgSeq); m_ackFlusher.ackOffsetMessage(proto); } } catch (DalException e) { log.error("Failed to ack messages(topic={}, partition={}, priority={}, groupId={}).", tpp.getTopic(), tpp.getPartition(), tpp.isPriority(), groupId, e); } }  @Override void appendMessages(String topicName, int partition, boolean priority,
	      Collection<MessageBatchWithRawData> batches); @Override synchronized Object findLastOffset(Tpp tpp, int groupId); @Override FetchResult fetchMessages(Tpp tpp, List<Object> offsets); @Override FetchResult fetchMessages(Tpp tpp, Object startOffset, int batchSize, String filter); @Override void nack(Tpp tpp, String groupId, boolean resend, List<Pair<Long, MessageMeta>> msgId2Metas); @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); @Override Object findMessageOffsetByTime(Tpp tpp, long time); @Override void initialize();  }