@Test public void sendCheckMessage() { listener.resolveDiscardMsg(createMessageExt()); }
@Override public void resolveDiscardMsg(MessageExt msgExt) { log.error("MsgExt:{} has been checked too many times, so discard it by moving it to system topic TRANS_CHECK_MAXTIME_TOPIC", msgExt); try { MessageExtBrokerInner brokerInner = toMessageExtBrokerInner(msgExt); PutMessageResult putMessageResult = this.getBrokerController().getMessageStore().putMessage(brokerInner); if (putMessageResult != null && putMessageResult.getPutMessageStatus() == PutMessageStatus.PUT_OK) { log.info("Put checked-too-many-time half message to TRANS_CHECK_MAXTIME_TOPIC OK. Restored in queueOffset={}, " + "commitLogOffset={}, real topic={}", msgExt.getQueueOffset(), msgExt.getCommitLogOffset(), msgExt.getUserProperty(MessageConst.PROPERTY_REAL_TOPIC)); } else { log.error("Put checked-too-many-time half message to TRANS_CHECK_MAXTIME_TOPIC failed, real topic={}, msgId={}", msgExt.getTopic(), msgExt.getMsgId()); } } catch (Exception e) { log.warn("Put checked-too-many-time message to TRANS_CHECK_MAXTIME_TOPIC error. {}", e); } }
DefaultTransactionalMessageCheckListener extends AbstractTransactionalMessageCheckListener { @Override public void resolveDiscardMsg(MessageExt msgExt) { log.error("MsgExt:{} has been checked too many times, so discard it by moving it to system topic TRANS_CHECK_MAXTIME_TOPIC", msgExt); try { MessageExtBrokerInner brokerInner = toMessageExtBrokerInner(msgExt); PutMessageResult putMessageResult = this.getBrokerController().getMessageStore().putMessage(brokerInner); if (putMessageResult != null && putMessageResult.getPutMessageStatus() == PutMessageStatus.PUT_OK) { log.info("Put checked-too-many-time half message to TRANS_CHECK_MAXTIME_TOPIC OK. Restored in queueOffset={}, " + "commitLogOffset={}, real topic={}", msgExt.getQueueOffset(), msgExt.getCommitLogOffset(), msgExt.getUserProperty(MessageConst.PROPERTY_REAL_TOPIC)); } else { log.error("Put checked-too-many-time half message to TRANS_CHECK_MAXTIME_TOPIC failed, real topic={}, msgId={}", msgExt.getTopic(), msgExt.getMsgId()); } } catch (Exception e) { log.warn("Put checked-too-many-time message to TRANS_CHECK_MAXTIME_TOPIC error. {}", e); } } }
DefaultTransactionalMessageCheckListener extends AbstractTransactionalMessageCheckListener { @Override public void resolveDiscardMsg(MessageExt msgExt) { log.error("MsgExt:{} has been checked too many times, so discard it by moving it to system topic TRANS_CHECK_MAXTIME_TOPIC", msgExt); try { MessageExtBrokerInner brokerInner = toMessageExtBrokerInner(msgExt); PutMessageResult putMessageResult = this.getBrokerController().getMessageStore().putMessage(brokerInner); if (putMessageResult != null && putMessageResult.getPutMessageStatus() == PutMessageStatus.PUT_OK) { log.info("Put checked-too-many-time half message to TRANS_CHECK_MAXTIME_TOPIC OK. Restored in queueOffset={}, " + "commitLogOffset={}, real topic={}", msgExt.getQueueOffset(), msgExt.getCommitLogOffset(), msgExt.getUserProperty(MessageConst.PROPERTY_REAL_TOPIC)); } else { log.error("Put checked-too-many-time half message to TRANS_CHECK_MAXTIME_TOPIC failed, real topic={}, msgId={}", msgExt.getTopic(), msgExt.getMsgId()); } } catch (Exception e) { log.warn("Put checked-too-many-time message to TRANS_CHECK_MAXTIME_TOPIC error. {}", e); } } DefaultTransactionalMessageCheckListener(); }
DefaultTransactionalMessageCheckListener extends AbstractTransactionalMessageCheckListener { @Override public void resolveDiscardMsg(MessageExt msgExt) { log.error("MsgExt:{} has been checked too many times, so discard it by moving it to system topic TRANS_CHECK_MAXTIME_TOPIC", msgExt); try { MessageExtBrokerInner brokerInner = toMessageExtBrokerInner(msgExt); PutMessageResult putMessageResult = this.getBrokerController().getMessageStore().putMessage(brokerInner); if (putMessageResult != null && putMessageResult.getPutMessageStatus() == PutMessageStatus.PUT_OK) { log.info("Put checked-too-many-time half message to TRANS_CHECK_MAXTIME_TOPIC OK. Restored in queueOffset={}, " + "commitLogOffset={}, real topic={}", msgExt.getQueueOffset(), msgExt.getCommitLogOffset(), msgExt.getUserProperty(MessageConst.PROPERTY_REAL_TOPIC)); } else { log.error("Put checked-too-many-time half message to TRANS_CHECK_MAXTIME_TOPIC failed, real topic={}, msgId={}", msgExt.getTopic(), msgExt.getMsgId()); } } catch (Exception e) { log.warn("Put checked-too-many-time message to TRANS_CHECK_MAXTIME_TOPIC error. {}", e); } } DefaultTransactionalMessageCheckListener(); @Override void resolveDiscardMsg(MessageExt msgExt); }
DefaultTransactionalMessageCheckListener extends AbstractTransactionalMessageCheckListener { @Override public void resolveDiscardMsg(MessageExt msgExt) { log.error("MsgExt:{} has been checked too many times, so discard it by moving it to system topic TRANS_CHECK_MAXTIME_TOPIC", msgExt); try { MessageExtBrokerInner brokerInner = toMessageExtBrokerInner(msgExt); PutMessageResult putMessageResult = this.getBrokerController().getMessageStore().putMessage(brokerInner); if (putMessageResult != null && putMessageResult.getPutMessageStatus() == PutMessageStatus.PUT_OK) { log.info("Put checked-too-many-time half message to TRANS_CHECK_MAXTIME_TOPIC OK. Restored in queueOffset={}, " + "commitLogOffset={}, real topic={}", msgExt.getQueueOffset(), msgExt.getCommitLogOffset(), msgExt.getUserProperty(MessageConst.PROPERTY_REAL_TOPIC)); } else { log.error("Put checked-too-many-time half message to TRANS_CHECK_MAXTIME_TOPIC failed, real topic={}, msgId={}", msgExt.getTopic(), msgExt.getMsgId()); } } catch (Exception e) { log.warn("Put checked-too-many-time message to TRANS_CHECK_MAXTIME_TOPIC error. {}", e); } } DefaultTransactionalMessageCheckListener(); @Override void resolveDiscardMsg(MessageExt msgExt);  }