@Test public void testAppendMessages() throws Exception { String topic = "order_new"; Collection<MessageBatchWithRawData> batches = new ArrayList<>(); ByteBuf rawData = Unpooled.buffer(); MessageBatchWithRawData batch = new MessageBatchWithRawData(topic, Arrays.asList(1), rawData, null); batches.add(batch); s.appendMessages(topic, 0, true, batches); }
@Override public void appendMessages(String topicName, int partition, boolean priority, Collection<MessageBatchWithRawData> batches) throws Exception { List<MessagePriority> msgs = new LinkedList<>(); Topic topic = m_metaService.findTopicByName(topicName); int count = 0; long bytes = 0; for (MessageBatchWithRawData batch : batches) { List<PartialDecodedMessage> pdmsgs = batch.getMessages(); BrokerStatusMonitor.INSTANCE.msgSaved(topicName, partition, pdmsgs.size()); for (PartialDecodedMessage pdmsg : pdmsgs) { count++; bytes += pdmsg.getBody().readableBytes() + pdmsg.getDurableProperties().readableBytes(); MessagePriority msg = new MessagePriority(); msg.setAttributes(pdmsg.readDurableProperties()); msg.setCreationDate(new Date(pdmsg.getBornTime())); msg.setPartition(partition); msg.setPayload(new ByteBufInputStream(pdmsg.getBody().duplicate())); if (topic.isPriorityMessageEnabled()) { msg.setPriority(priority ? 0 : 1); } else { msg.setPriority(1); } msg.setProducerId(0); msg.setProducerIp(""); msg.setRefKey(pdmsg.getKey()); msg.setTopic(topicName); msg.setCodecType(pdmsg.getBodyCodecType()); msgs.add(msg); } } if (!msgs.isEmpty()) { batchInsert(topic, partition, priority, msgs); } if (count > 0) { logSelecotrMetric(topicName, partition, priority, count); } if (bytes > 0) { try { CatUtil.logEventPeriodically( CatConstants.TYPE_MESSAGE_BROKER_PRODUCE_BYTES + topic.getPartitions().get(partition).getWriteDatasource(), topicName, bytes); } catch (Exception e) { log.warn("Exception occurred while loging bytes for {}-{}", topicName, partition, e); } } }
MySQLMessageQueueStorage implements MessageQueueStorage, Initializable { @Override public void appendMessages(String topicName, int partition, boolean priority, Collection<MessageBatchWithRawData> batches) throws Exception { List<MessagePriority> msgs = new LinkedList<>(); Topic topic = m_metaService.findTopicByName(topicName); int count = 0; long bytes = 0; for (MessageBatchWithRawData batch : batches) { List<PartialDecodedMessage> pdmsgs = batch.getMessages(); BrokerStatusMonitor.INSTANCE.msgSaved(topicName, partition, pdmsgs.size()); for (PartialDecodedMessage pdmsg : pdmsgs) { count++; bytes += pdmsg.getBody().readableBytes() + pdmsg.getDurableProperties().readableBytes(); MessagePriority msg = new MessagePriority(); msg.setAttributes(pdmsg.readDurableProperties()); msg.setCreationDate(new Date(pdmsg.getBornTime())); msg.setPartition(partition); msg.setPayload(new ByteBufInputStream(pdmsg.getBody().duplicate())); if (topic.isPriorityMessageEnabled()) { msg.setPriority(priority ? 0 : 1); } else { msg.setPriority(1); } msg.setProducerId(0); msg.setProducerIp(""); msg.setRefKey(pdmsg.getKey()); msg.setTopic(topicName); msg.setCodecType(pdmsg.getBodyCodecType()); msgs.add(msg); } } if (!msgs.isEmpty()) { batchInsert(topic, partition, priority, msgs); } if (count > 0) { logSelecotrMetric(topicName, partition, priority, count); } if (bytes > 0) { try { CatUtil.logEventPeriodically( CatConstants.TYPE_MESSAGE_BROKER_PRODUCE_BYTES + topic.getPartitions().get(partition).getWriteDatasource(), topicName, bytes); } catch (Exception e) { log.warn("Exception occurred while loging bytes for {}-{}", topicName, partition, e); } } } }
MySQLMessageQueueStorage implements MessageQueueStorage, Initializable { @Override public void appendMessages(String topicName, int partition, boolean priority, Collection<MessageBatchWithRawData> batches) throws Exception { List<MessagePriority> msgs = new LinkedList<>(); Topic topic = m_metaService.findTopicByName(topicName); int count = 0; long bytes = 0; for (MessageBatchWithRawData batch : batches) { List<PartialDecodedMessage> pdmsgs = batch.getMessages(); BrokerStatusMonitor.INSTANCE.msgSaved(topicName, partition, pdmsgs.size()); for (PartialDecodedMessage pdmsg : pdmsgs) { count++; bytes += pdmsg.getBody().readableBytes() + pdmsg.getDurableProperties().readableBytes(); MessagePriority msg = new MessagePriority(); msg.setAttributes(pdmsg.readDurableProperties()); msg.setCreationDate(new Date(pdmsg.getBornTime())); msg.setPartition(partition); msg.setPayload(new ByteBufInputStream(pdmsg.getBody().duplicate())); if (topic.isPriorityMessageEnabled()) { msg.setPriority(priority ? 0 : 1); } else { msg.setPriority(1); } msg.setProducerId(0); msg.setProducerIp(""); msg.setRefKey(pdmsg.getKey()); msg.setTopic(topicName); msg.setCodecType(pdmsg.getBodyCodecType()); msgs.add(msg); } } if (!msgs.isEmpty()) { batchInsert(topic, partition, priority, msgs); } if (count > 0) { logSelecotrMetric(topicName, partition, priority, count); } if (bytes > 0) { try { CatUtil.logEventPeriodically( CatConstants.TYPE_MESSAGE_BROKER_PRODUCE_BYTES + topic.getPartitions().get(partition).getWriteDatasource(), topicName, bytes); } catch (Exception e) { log.warn("Exception occurred while loging bytes for {}-{}", topicName, partition, e); } } }  }
MySQLMessageQueueStorage implements MessageQueueStorage, Initializable { @Override public void appendMessages(String topicName, int partition, boolean priority, Collection<MessageBatchWithRawData> batches) throws Exception { List<MessagePriority> msgs = new LinkedList<>(); Topic topic = m_metaService.findTopicByName(topicName); int count = 0; long bytes = 0; for (MessageBatchWithRawData batch : batches) { List<PartialDecodedMessage> pdmsgs = batch.getMessages(); BrokerStatusMonitor.INSTANCE.msgSaved(topicName, partition, pdmsgs.size()); for (PartialDecodedMessage pdmsg : pdmsgs) { count++; bytes += pdmsg.getBody().readableBytes() + pdmsg.getDurableProperties().readableBytes(); MessagePriority msg = new MessagePriority(); msg.setAttributes(pdmsg.readDurableProperties()); msg.setCreationDate(new Date(pdmsg.getBornTime())); msg.setPartition(partition); msg.setPayload(new ByteBufInputStream(pdmsg.getBody().duplicate())); if (topic.isPriorityMessageEnabled()) { msg.setPriority(priority ? 0 : 1); } else { msg.setPriority(1); } msg.setProducerId(0); msg.setProducerIp(""); msg.setRefKey(pdmsg.getKey()); msg.setTopic(topicName); msg.setCodecType(pdmsg.getBodyCodecType()); msgs.add(msg); } } if (!msgs.isEmpty()) { batchInsert(topic, partition, priority, msgs); } if (count > 0) { logSelecotrMetric(topicName, partition, priority, count); } if (bytes > 0) { try { CatUtil.logEventPeriodically( CatConstants.TYPE_MESSAGE_BROKER_PRODUCE_BYTES + topic.getPartitions().get(partition).getWriteDatasource(), topicName, bytes); } catch (Exception e) { log.warn("Exception occurred while loging bytes for {}-{}", topicName, partition, 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 appendMessages(String topicName, int partition, boolean priority, Collection<MessageBatchWithRawData> batches) throws Exception { List<MessagePriority> msgs = new LinkedList<>(); Topic topic = m_metaService.findTopicByName(topicName); int count = 0; long bytes = 0; for (MessageBatchWithRawData batch : batches) { List<PartialDecodedMessage> pdmsgs = batch.getMessages(); BrokerStatusMonitor.INSTANCE.msgSaved(topicName, partition, pdmsgs.size()); for (PartialDecodedMessage pdmsg : pdmsgs) { count++; bytes += pdmsg.getBody().readableBytes() + pdmsg.getDurableProperties().readableBytes(); MessagePriority msg = new MessagePriority(); msg.setAttributes(pdmsg.readDurableProperties()); msg.setCreationDate(new Date(pdmsg.getBornTime())); msg.setPartition(partition); msg.setPayload(new ByteBufInputStream(pdmsg.getBody().duplicate())); if (topic.isPriorityMessageEnabled()) { msg.setPriority(priority ? 0 : 1); } else { msg.setPriority(1); } msg.setProducerId(0); msg.setProducerIp(""); msg.setRefKey(pdmsg.getKey()); msg.setTopic(topicName); msg.setCodecType(pdmsg.getBodyCodecType()); msgs.add(msg); } } if (!msgs.isEmpty()) { batchInsert(topic, partition, priority, msgs); } if (count > 0) { logSelecotrMetric(topicName, partition, priority, count); } if (bytes > 0) { try { CatUtil.logEventPeriodically( CatConstants.TYPE_MESSAGE_BROKER_PRODUCE_BYTES + topic.getPartitions().get(partition).getWriteDatasource(), topicName, bytes); } catch (Exception e) { log.warn("Exception occurred while loging bytes for {}-{}", topicName, partition, 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();  }