apache / incubator-rocketmq / 654

Build # 654

Build Type

push

travis-ci

Committed by yukon

Commit Message

<a href="https://github.com/apache/incubator-rocketmq/commit/<a class=hub.com/apache/incubator-rocketmq/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/incubator-rocketmq/commit/<a class="double-link" href="https://git"><a class=hub.com/apache/incubator-rocketmq/commit/d4149207e27ed3516f1f06407b55986790b806ae">d4149207e">&lt;a href=&quot;https://github.com/apache/incubator-rocketmq/commit/</a><a class="double-link" href="https://github.com/apache/incubator-rocketmq/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/incubator-rocketmq/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/incubator-rocketmq/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/incubator-rocketmq/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/incubator-rocketmq/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/incubator-rocketmq/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/incubator-rocketmq/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/incubator-rocketmq/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/incubator-rocketmq/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/incubator-rocketmq/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/apache/incubator-rocketmq/commit/d4149207e27ed3516f1f06407b55986790b806ae">d4149207e</a><a href="https://github.com/apache/incubator-rocketmq/commit/d4149207e27ed3516f1f06407b55986790b806ae">&lt;a href=&quot;https://github.com/apache/incubator-rocketmq/commit/d4149207e27ed3516f1f06407b55986790b806ae&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/apache/incubator-rocketmq/commit/&amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/apache/incubator-rocketmq/commit/&amp;amp;lt;a class=&amp;amp;quot;double-link... (continued)

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

10467 of 26754 relevant lines covered (39.12%)

1.13 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

39.11

/store/src/main/java/org/apache/rocketmq/store/schedule/ScheduleMessageService.java

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.rocketmq.store.schedule;

import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.apache.rocketmq.common.ConfigManager;
import org.apache.rocketmq.common.TopicFilterType;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.message.MessageAccessor;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageDecoder;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.running.RunningStats;
import org.apache.rocketmq.store.ConsumeQueue;
import org.apache.rocketmq.store.ConsumeQueueExt;
import org.apache.rocketmq.store.DefaultMessageStore;
import org.apache.rocketmq.store.MessageExtBrokerInner;
import org.apache.rocketmq.store.PutMessageResult;
import org.apache.rocketmq.store.PutMessageStatus;
import org.apache.rocketmq.store.SelectMappedBufferResult;
import org.apache.rocketmq.store.config.StorePathConfigHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ScheduleMessageService extends ConfigManager {
    public static final String SCHEDULE_TOPIC = "SCHEDULE_TOPIC_XXXX";
    private static final Logger log = LoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);
    private static final long FIRST_DELAY_TIME = 1000L;
    private static final long DELAY_FOR_A_WHILE = 100L;
    private static final long DELAY_FOR_A_PERIOD = 10000L;

    private final ConcurrentMap<Integer /* level */, Long/* delay timeMillis */> delayLevelTable =
        new ConcurrentHashMap<Integer, Long>(32);

    private final ConcurrentMap<Integer /* level */, Long/* offset */> offsetTable =
        new ConcurrentHashMap<Integer, Long>(32);

    private final Timer timer = new Timer("ScheduleMessageTimerThread", true);

    private final DefaultMessageStore defaultMessageStore;

    private int maxDelayLevel;

    public ScheduleMessageService(final DefaultMessageStore defaultMessageStore) {
        this.defaultMessageStore = defaultMessageStore;
    }

    public static int queueId2DelayLevel(final int queueId) {
        return queueId + 1;
    }

    public static int delayLevel2QueueId(final int delayLevel) {
        return delayLevel - 1;
    }

    public void buildRunningStats(HashMap<String, String> stats) {
        Iterator<Entry<Integer, Long>> it = this.offsetTable.entrySet().iterator();
        while (it.hasNext()) {
            Entry<Integer, Long> next = it.next();
            int queueId = delayLevel2QueueId(next.getKey());
            long delayOffset = next.getValue();
            long maxOffset = this.defaultMessageStore.getMaxOffsetInQueue(SCHEDULE_TOPIC, queueId);
            String value = String.format("%d,%d", delayOffset, maxOffset);
            String key = String.format("%s_%d", RunningStats.scheduleMessageOffset.name(), next.getKey());
            stats.put(key, value);
        }
    }

    private void updateOffset(int delayLevel, long offset) {
        this.offsetTable.put(delayLevel, offset);
    }

    public long computeDeliverTimestamp(final int delayLevel, final long storeTimestamp) {
        Long time = this.delayLevelTable.get(delayLevel);
        if (time != null) {
            return time + storeTimestamp;
        }

        return storeTimestamp + 1000;
    }

    public void start() {

        for (Map.Entry<Integer, Long> entry : this.delayLevelTable.entrySet()) {
            Integer level = entry.getKey();
            Long timeDelay = entry.getValue();
            Long offset = this.offsetTable.get(level);
            if (null == offset) {
                offset = 0L;
            }

            if (timeDelay != null) {
                this.timer.schedule(new DeliverDelayedMessageTimerTask(level, offset), FIRST_DELAY_TIME);
            }
        }

        this.timer.scheduleAtFixedRate(new TimerTask() {

            @Override
            public void run() {
                try {
                    ScheduleMessageService.this.persist();
                } catch (Exception e) {
                    log.error("scheduleAtFixedRate flush exception", e);
                }
            }
        }, 10000, this.defaultMessageStore.getMessageStoreConfig().getFlushDelayOffsetInterval());
    }

    public void shutdown() {
        this.timer.cancel();
    }

    public int getMaxDelayLevel() {
        return maxDelayLevel;
    }

    public String encode() {
        return this.encode(false);
    }

    public boolean load() {
        boolean result = super.load();
        result = result && this.parseDelayLevel();
        return result;
    }

    @Override
    public String configFilePath() {
        return StorePathConfigHelper.getDelayOffsetStorePath(this.defaultMessageStore.getMessageStoreConfig()
            .getStorePathRootDir());
    }

    @Override
    public void decode(String jsonString) {
        if (jsonString != null) {
            DelayOffsetSerializeWrapper delayOffsetSerializeWrapper =
                DelayOffsetSerializeWrapper.fromJson(jsonString, DelayOffsetSerializeWrapper.class);
            if (delayOffsetSerializeWrapper != null) {
                this.offsetTable.putAll(delayOffsetSerializeWrapper.getOffsetTable());
            }
        }
    }

    public String encode(final boolean prettyFormat) {
        DelayOffsetSerializeWrapper delayOffsetSerializeWrapper = new DelayOffsetSerializeWrapper();
        delayOffsetSerializeWrapper.setOffsetTable(this.offsetTable);
        return delayOffsetSerializeWrapper.toJson(prettyFormat);
    }

    public boolean parseDelayLevel() {
        HashMap<String, Long> timeUnitTable = new HashMap<String, Long>();
        timeUnitTable.put("s", 1000L);
        timeUnitTable.put("m", 1000L * 60);
        timeUnitTable.put("h", 1000L * 60 * 60);
        timeUnitTable.put("d", 1000L * 60 * 60 * 24);

        String levelString = this.defaultMessageStore.getMessageStoreConfig().getMessageDelayLevel();
        try {
            String[] levelArray = levelString.split(" ");
            for (int i = 0; i < levelArray.length; i++) {
                String value = levelArray[i];
                String ch = value.substring(value.length() - 1);
                Long tu = timeUnitTable.get(ch);

                int level = i + 1;
                if (level > this.maxDelayLevel) {
                    this.maxDelayLevel = level;
                }
                long num = Long.parseLong(value.substring(0, value.length() - 1));
                long delayTimeMillis = tu * num;
                this.delayLevelTable.put(level, delayTimeMillis);
            }
        } catch (Exception e) {
            log.error("parseDelayLevel exception", e);
            log.info("levelString String = {}", levelString);
            return false;
        }

        return true;
    }

    class DeliverDelayedMessageTimerTask extends TimerTask {
        private final int delayLevel;
        private final long offset;

        public DeliverDelayedMessageTimerTask(int delayLevel, long offset) {
            this.delayLevel = delayLevel;
            this.offset = offset;
        }

        @Override
        public void run() {
            try {
                this.executeOnTimeup();
            } catch (Exception e) {
                // XXX: warn and notify me
                log.error("ScheduleMessageService, executeOnTimeup exception", e);
                ScheduleMessageService.this.timer.schedule(new DeliverDelayedMessageTimerTask(
                    this.delayLevel, this.offset), DELAY_FOR_A_PERIOD);
            }
        }

        /**
         * @return
         */
        private long correctDeliverTimestamp(final long now, final long deliverTimestamp) {

            long result = deliverTimestamp;

            long maxTimestamp = now + ScheduleMessageService.this.delayLevelTable.get(this.delayLevel);
            if (deliverTimestamp > maxTimestamp) {
                result = now;
            }

            return result;
        }

        public void executeOnTimeup() {
            ConsumeQueue cq =
                ScheduleMessageService.this.defaultMessageStore.findConsumeQueue(SCHEDULE_TOPIC,
                    delayLevel2QueueId(delayLevel));

            long failScheduleOffset = offset;

            if (cq != null) {
                SelectMappedBufferResult bufferCQ = cq.getIndexBuffer(this.offset);
                if (bufferCQ != null) {
                    try {
                        long nextOffset = offset;
                        int i = 0;
                        ConsumeQueueExt.CqExtUnit cqExtUnit = new ConsumeQueueExt.CqExtUnit();
                        for (; i < bufferCQ.getSize(); i += ConsumeQueue.CQ_STORE_UNIT_SIZE) {
                            long offsetPy = bufferCQ.getByteBuffer().getLong();
                            int sizePy = bufferCQ.getByteBuffer().getInt();
                            long tagsCode = bufferCQ.getByteBuffer().getLong();

                            if (cq.isExtAddr(tagsCode)) {
                                if (cq.getExt(tagsCode, cqExtUnit)) {
                                    tagsCode = cqExtUnit.getTagsCode();
                                } else {
                                    //can't find ext content.So re compute tags code.
                                    log.error("[BUG] can't find consume queue extend file content!addr={}, offsetPy={}, sizePy={}",
                                        tagsCode, offsetPy, sizePy);
                                    long msgStoreTime = defaultMessageStore.getCommitLog().pickupStoreTimestamp(offsetPy, sizePy);
                                    tagsCode = computeDeliverTimestamp(delayLevel, msgStoreTime);
                                }
                            }

                            long now = System.currentTimeMillis();
                            long deliverTimestamp = this.correctDeliverTimestamp(now, tagsCode);

                            nextOffset = offset + (i / ConsumeQueue.CQ_STORE_UNIT_SIZE);

                            long countdown = deliverTimestamp - now;

                            if (countdown <= 0) {
                                MessageExt msgExt =
                                    ScheduleMessageService.this.defaultMessageStore.lookMessageByOffset(
                                        offsetPy, sizePy);

                                if (msgExt != null) {
                                    try {
                                        MessageExtBrokerInner msgInner = this.messageTimeup(msgExt);
                                        PutMessageResult putMessageResult =
                                            ScheduleMessageService.this.defaultMessageStore
                                                .putMessage(msgInner);

                                        if (putMessageResult != null
                                            && putMessageResult.getPutMessageStatus() == PutMessageStatus.PUT_OK) {
                                            continue;
                                        } else {
                                            // XXX: warn and notify me
                                            log.error(
                                                "ScheduleMessageService, a message time up, but reput it failed, topic: {} msgId {}",
                                                msgExt.getTopic(), msgExt.getMsgId());
                                            ScheduleMessageService.this.timer.schedule(
                                                new DeliverDelayedMessageTimerTask(this.delayLevel,
                                                    nextOffset), DELAY_FOR_A_PERIOD);
                                            ScheduleMessageService.this.updateOffset(this.delayLevel,
                                                nextOffset);
                                            return;
                                        }
                                    } catch (Exception e) {
                                        /*
                                         * XXX: warn and notify me



                                         */
                                        log.error(
                                            "ScheduleMessageService, messageTimeup execute error, drop it. msgExt="
                                                + msgExt + ", nextOffset=" + nextOffset + ",offsetPy="
                                                + offsetPy + ",sizePy=" + sizePy, e);
                                    }
                                }
                            } else {
                                ScheduleMessageService.this.timer.schedule(
                                    new DeliverDelayedMessageTimerTask(this.delayLevel, nextOffset),
                                    countdown);
                                ScheduleMessageService.this.updateOffset(this.delayLevel, nextOffset);
                                return;
                            }
                        } // end of for

                        nextOffset = offset + (i / ConsumeQueue.CQ_STORE_UNIT_SIZE);
                        ScheduleMessageService.this.timer.schedule(new DeliverDelayedMessageTimerTask(
                            this.delayLevel, nextOffset), DELAY_FOR_A_WHILE);
                        ScheduleMessageService.this.updateOffset(this.delayLevel, nextOffset);
                        return;
                    } finally {

                        bufferCQ.release();
                    }
                } // end of if (bufferCQ != null)
                else {
                    /*


                     */
                    long cqMinOffset = cq.getMinOffsetInQueue();
                    if (offset < cqMinOffset) {
                        failScheduleOffset = cqMinOffset;
                        log.error("schedule CQ offset invalid. offset=" + offset + ", cqMinOffset="
                            + cqMinOffset + ", queueId=" + cq.getQueueId());
                    }
                }
            } // end of if (cq != null)

            ScheduleMessageService.this.timer.schedule(new DeliverDelayedMessageTimerTask(this.delayLevel,
                failScheduleOffset), DELAY_FOR_A_WHILE);
        }

        private MessageExtBrokerInner messageTimeup(MessageExt msgExt) {
            MessageExtBrokerInner msgInner = new MessageExtBrokerInner();
            msgInner.setBody(msgExt.getBody());
            msgInner.setFlag(msgExt.getFlag());
            MessageAccessor.setProperties(msgInner, msgExt.getProperties());

            TopicFilterType topicFilterType = MessageExt.parseTopicFilterType(msgInner.getSysFlag());
            long tagsCodeValue =
                MessageExtBrokerInner.tagsString2tagsCode(topicFilterType, msgInner.getTags());
            msgInner.setTagsCode(tagsCodeValue);
            msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgExt.getProperties()));

            msgInner.setSysFlag(msgExt.getSysFlag());
            msgInner.setBornTimestamp(msgExt.getBornTimestamp());
            msgInner.setBornHost(msgExt.getBornHost());
            msgInner.setStoreHost(msgExt.getStoreHost());
            msgInner.setReconsumeTimes(msgExt.getReconsumeTimes());

            msgInner.setWaitStoreMsgOK(false);
            MessageAccessor.clearProperty(msgInner, MessageConst.PROPERTY_DELAY_TIME_LEVEL);

            msgInner.setTopic(msgInner.getProperty(MessageConst.PROPERTY_REAL_TOPIC));

            String queueIdStr = msgInner.getProperty(MessageConst.PROPERTY_REAL_QUEUE_ID);
            int queueId = Integer.parseInt(queueIdStr);
            msgInner.setQueueId(queueId);

            return msgInner;
        }
    }
}

1	/*
2	* Licensed to the Apache Software Foundation (ASF) under one or more
3	* contributor license agreements. See the NOTICE file distributed with
4	* this work for additional information regarding copyright ownership.
5	* The ASF licenses this file to You under the Apache License, Version 2.0
6	* (the "License"); you may not use this file except in compliance with
7	* the License. You may obtain a copy of the License at
8	*
9	* http://www.apache.org/licenses/LICENSE-2.0
10	*
11	* Unless required by applicable law or agreed to in writing, software
12	* distributed under the License is distributed on an "AS IS" BASIS,
13	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	* See the License for the specific language governing permissions and
15	* limitations under the License.
16	*/
17	package org.apache.rocketmq.store.schedule;
18
19	import java.util.HashMap;
20	import java.util.Iterator;
21	import java.util.Map;
22	import java.util.Map.Entry;
23	import java.util.Timer;
24	import java.util.TimerTask;
25	import java.util.concurrent.ConcurrentHashMap;
26	import java.util.concurrent.ConcurrentMap;
27	import org.apache.rocketmq.common.ConfigManager;
28	import org.apache.rocketmq.common.TopicFilterType;
29	import org.apache.rocketmq.common.constant.LoggerName;
30	import org.apache.rocketmq.common.message.MessageAccessor;
31	import org.apache.rocketmq.common.message.MessageConst;
32	import org.apache.rocketmq.common.message.MessageDecoder;
33	import org.apache.rocketmq.common.message.MessageExt;
34	import org.apache.rocketmq.common.running.RunningStats;
35	import org.apache.rocketmq.store.ConsumeQueue;
36	import org.apache.rocketmq.store.ConsumeQueueExt;
37	import org.apache.rocketmq.store.DefaultMessageStore;
38	import org.apache.rocketmq.store.MessageExtBrokerInner;
39	import org.apache.rocketmq.store.PutMessageResult;
40	import org.apache.rocketmq.store.PutMessageStatus;
41	import org.apache.rocketmq.store.SelectMappedBufferResult;
42	import org.apache.rocketmq.store.config.StorePathConfigHelper;
43	import org.slf4j.Logger;
44	import org.slf4j.LoggerFactory;
45
46	public class ScheduleMessageService extends ConfigManager {
47	public static final String SCHEDULE_TOPIC = "SCHEDULE_TOPIC_XXXX";
48	private static final Logger log = LoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);	3✔
49	private static final long FIRST_DELAY_TIME = 1000L;
50	private static final long DELAY_FOR_A_WHILE = 100L;
51	private static final long DELAY_FOR_A_PERIOD = 10000L;
52
53	private final ConcurrentMap<Integer /* level /, Long/ delay timeMillis */> delayLevelTable =	3✔
54	new ConcurrentHashMap<Integer, Long>(32);
55
56	private final ConcurrentMap<Integer /* level /, Long/ offset */> offsetTable =	3✔
57	new ConcurrentHashMap<Integer, Long>(32);
58
59	private final Timer timer = new Timer("ScheduleMessageTimerThread", true);	3✔
60
61	private final DefaultMessageStore defaultMessageStore;
62
63	private int maxDelayLevel;
64
65	public ScheduleMessageService(final DefaultMessageStore defaultMessageStore) {	3✔
66	this.defaultMessageStore = defaultMessageStore;	3✔
67	}	3✔
68
69	public static int queueId2DelayLevel(final int queueId) {
70	return queueId + 1;	×
71	}
72
73	public static int delayLevel2QueueId(final int delayLevel) {
74	return delayLevel - 1;	3✔
75	}
76
77	public void buildRunningStats(HashMap<String, String> stats) {
78	Iterator<Entry<Integer, Long>> it = this.offsetTable.entrySet().iterator();	×
79	while (it.hasNext()) {	×
80	Entry<Integer, Long> next = it.next();	×
81	int queueId = delayLevel2QueueId(next.getKey());	×
82	long delayOffset = next.getValue();	×
83	long maxOffset = this.defaultMessageStore.getMaxOffsetInQueue(SCHEDULE_TOPIC, queueId);	×
84	String value = String.format("%d,%d", delayOffset, maxOffset);	×
85	String key = String.format("%s_%d", RunningStats.scheduleMessageOffset.name(), next.getKey());	×
86	stats.put(key, value);	×
87	}	×
88	}	×
89
90	private void updateOffset(int delayLevel, long offset) {
91	this.offsetTable.put(delayLevel, offset);	×
92	}	×
93
94	public long computeDeliverTimestamp(final int delayLevel, final long storeTimestamp) {
95	Long time = this.delayLevelTable.get(delayLevel);	×
96	if (time != null) {	×
97	return time + storeTimestamp;	×
98	}
99
100	return storeTimestamp + 1000;	×
101	}
102
103	public void start() {
104
105	for (Map.Entry<Integer, Long> entry : this.delayLevelTable.entrySet()) {	3✔
106	Integer level = entry.getKey();	3✔
107	Long timeDelay = entry.getValue();	3✔
108	Long offset = this.offsetTable.get(level);	3✔
109	if (null == offset) {	3✔
110	offset = 0L;	3✔
111	}
112
113	if (timeDelay != null) {	3✔
114	this.timer.schedule(new DeliverDelayedMessageTimerTask(level, offset), FIRST_DELAY_TIME);	3✔
115	}
116	}	3✔
117
118	this.timer.scheduleAtFixedRate(new TimerTask() {	3✔
119
120	@Override
121	public void run() {
122	try {
123	ScheduleMessageService.this.persist();	×
124	} catch (Exception e) {	×
125	log.error("scheduleAtFixedRate flush exception", e);	×
126	}	×
127	}	×
128	}, 10000, this.defaultMessageStore.getMessageStoreConfig().getFlushDelayOffsetInterval());	1 only CUSTOM_JDK="oraclejdk8" ✔
129	}	3✔
130
131	public void shutdown() {
132	this.timer.cancel();	3✔
133	}	3✔
134
135	public int getMaxDelayLevel() {
136	return maxDelayLevel;	×
137	}
138
139	public String encode() {
140	return this.encode(false);	×
141	}
142
143	public boolean load() {
144	boolean result = super.load();	3✔
145	result = result && this.parseDelayLevel();	3✔
146	return result;	3✔
147	}
148
149	@Override
150	public String configFilePath() {
151	return StorePathConfigHelper.getDelayOffsetStorePath(this.defaultMessageStore.getMessageStoreConfig()	3✔
152	.getStorePathRootDir());	1 only CUSTOM_JDK="oraclejdk8" ✔
153	}
154
155	@Override
156	public void decode(String jsonString) {
157	if (jsonString != null) {	1 only CUSTOM_JDK="oraclejdk8" ✔
158	DelayOffsetSerializeWrapper delayOffsetSerializeWrapper =	1 only CUSTOM_JDK="oraclejdk8" ✔
159	DelayOffsetSerializeWrapper.fromJson(jsonString, DelayOffsetSerializeWrapper.class);	1 only CUSTOM_JDK="oraclejdk8" ✔
160	if (delayOffsetSerializeWrapper != null) {	1 only CUSTOM_JDK="oraclejdk8" ✔
161	this.offsetTable.putAll(delayOffsetSerializeWrapper.getOffsetTable());	1 only CUSTOM_JDK="oraclejdk8" ✔
162	}
163	}
164	}	1 only CUSTOM_JDK="oraclejdk8" ✔
165
166	public String encode(final boolean prettyFormat) {
167	DelayOffsetSerializeWrapper delayOffsetSerializeWrapper = new DelayOffsetSerializeWrapper();	×
168	delayOffsetSerializeWrapper.setOffsetTable(this.offsetTable);	×
169	return delayOffsetSerializeWrapper.toJson(prettyFormat);	×
170	}
171
172	public boolean parseDelayLevel() {
173	HashMap<String, Long> timeUnitTable = new HashMap<String, Long>();	3✔
174	timeUnitTable.put("s", 1000L);	3✔
175	timeUnitTable.put("m", 1000L * 60);	3✔
176	timeUnitTable.put("h", 1000L * 60 * 60);	3✔
177	timeUnitTable.put("d", 1000L * 60 * 60 * 24);	3✔
178
179	String levelString = this.defaultMessageStore.getMessageStoreConfig().getMessageDelayLevel();	3✔
180	try {
181	String[] levelArray = levelString.split(" ");	3✔
182	for (int i = 0; i < levelArray.length; i++) {	3✔
183	String value = levelArray[i];	3✔
184	String ch = value.substring(value.length() - 1);	3✔
185	Long tu = timeUnitTable.get(ch);	3✔
186
187	int level = i + 1;	3✔
188	if (level > this.maxDelayLevel) {	3✔
189	this.maxDelayLevel = level;	3✔
190	}
191	long num = Long.parseLong(value.substring(0, value.length() - 1));	3✔
192	long delayTimeMillis = tu * num;	3✔
193	this.delayLevelTable.put(level, delayTimeMillis);	3✔
194	}
195	} catch (Exception e) {	×
196	log.error("parseDelayLevel exception", e);	×
197	log.info("levelString String = {}", levelString);	×
198	return false;	×
199	}	3✔
200
201	return true;	3✔
202	}
203
204	class DeliverDelayedMessageTimerTask extends TimerTask {
205	private final int delayLevel;
206	private final long offset;
207
208	public DeliverDelayedMessageTimerTask(int delayLevel, long offset) {	3✔
209	this.delayLevel = delayLevel;	3✔
210	this.offset = offset;	3✔
211	}	3✔
212
213	@Override
214	public void run() {
215	try {
216	this.executeOnTimeup();	3✔
217	} catch (Exception e) {	×
218	// XXX: warn and notify me
219	log.error("ScheduleMessageService, executeOnTimeup exception", e);	×
220	ScheduleMessageService.this.timer.schedule(new DeliverDelayedMessageTimerTask(	×
221	this.delayLevel, this.offset), DELAY_FOR_A_PERIOD);
222	}	3✔
223	}	3✔
224
225	/**
226	* @return
227	*/
228	private long correctDeliverTimestamp(final long now, final long deliverTimestamp) {
229
230	long result = deliverTimestamp;	×
231
232	long maxTimestamp = now + ScheduleMessageService.this.delayLevelTable.get(this.delayLevel);	×
233	if (deliverTimestamp > maxTimestamp) {	×
234	result = now;	×
235	}
236
237	return result;	×
238	}
239
240	public void executeOnTimeup() {
241	ConsumeQueue cq =	3✔
242	ScheduleMessageService.this.defaultMessageStore.findConsumeQueue(SCHEDULE_TOPIC,	1 only CUSTOM_JDK="oraclejdk8" ✔
243	delayLevel2QueueId(delayLevel));	1 only CUSTOM_JDK="oraclejdk8" ✔
244
245	long failScheduleOffset = offset;	3✔
246
247	if (cq != null) {	3✔
248	SelectMappedBufferResult bufferCQ = cq.getIndexBuffer(this.offset);	3✔
249	if (bufferCQ != null) {	3✔
250	try {
251	long nextOffset = offset;	×
252	int i = 0;	×
253	ConsumeQueueExt.CqExtUnit cqExtUnit = new ConsumeQueueExt.CqExtUnit();	×
254	for (; i < bufferCQ.getSize(); i += ConsumeQueue.CQ_STORE_UNIT_SIZE) {	×
255	long offsetPy = bufferCQ.getByteBuffer().getLong();	×
256	int sizePy = bufferCQ.getByteBuffer().getInt();	×
257	long tagsCode = bufferCQ.getByteBuffer().getLong();	×
258
259	if (cq.isExtAddr(tagsCode)) {	×
260	if (cq.getExt(tagsCode, cqExtUnit)) {	×
261	tagsCode = cqExtUnit.getTagsCode();	×
262	} else {
263	//can't find ext content.So re compute tags code.
264	log.error("[BUG] can't find consume queue extend file content!addr={}, offsetPy={}, sizePy={}",	×
265	tagsCode, offsetPy, sizePy);	×
266	long msgStoreTime = defaultMessageStore.getCommitLog().pickupStoreTimestamp(offsetPy, sizePy);	×
267	tagsCode = computeDeliverTimestamp(delayLevel, msgStoreTime);	×
268	}
269	}
270
271	long now = System.currentTimeMillis();	×
272	long deliverTimestamp = this.correctDeliverTimestamp(now, tagsCode);	×
273
274	nextOffset = offset + (i / ConsumeQueue.CQ_STORE_UNIT_SIZE);	×
275
276	long countdown = deliverTimestamp - now;	×
277
278	if (countdown <= 0) {	×
279	MessageExt msgExt =	×
280	ScheduleMessageService.this.defaultMessageStore.lookMessageByOffset(	×
281	offsetPy, sizePy);
282
283	if (msgExt != null) {	×
284	try {
285	MessageExtBrokerInner msgInner = this.messageTimeup(msgExt);	×
286	PutMessageResult putMessageResult =	×
287	ScheduleMessageService.this.defaultMessageStore	×
288	.putMessage(msgInner);	×
289
290	if (putMessageResult != null	×
291	&& putMessageResult.getPutMessageStatus() == PutMessageStatus.PUT_OK) {	×
292	continue;	×
293	} else {
294	// XXX: warn and notify me
295	log.error(	×
296	"ScheduleMessageService, a message time up, but reput it failed, topic: {} msgId {}",
297	msgExt.getTopic(), msgExt.getMsgId());	×
298	ScheduleMessageService.this.timer.schedule(	×
299	new DeliverDelayedMessageTimerTask(this.delayLevel,
300	nextOffset), DELAY_FOR_A_PERIOD);
301	ScheduleMessageService.this.updateOffset(this.delayLevel,	×
302	nextOffset);
303	return;	×
304	}
305	} catch (Exception e) {	×
306	/*
307	* XXX: warn and notify me
308
309
310
311	*/
312	log.error(	×
313	"ScheduleMessageService, messageTimeup execute error, drop it. msgExt="
314	+ msgExt + ", nextOffset=" + nextOffset + ",offsetPy="
315	+ offsetPy + ",sizePy=" + sizePy, e);
316	}
317	}
318	} else {	×
319	ScheduleMessageService.this.timer.schedule(	×
320	new DeliverDelayedMessageTimerTask(this.delayLevel, nextOffset),
321	countdown);
322	ScheduleMessageService.this.updateOffset(this.delayLevel, nextOffset);	×
323	return;	×
324	}
325	} // end of for
326
327	nextOffset = offset + (i / ConsumeQueue.CQ_STORE_UNIT_SIZE);	×
328	ScheduleMessageService.this.timer.schedule(new DeliverDelayedMessageTimerTask(	×
329	this.delayLevel, nextOffset), DELAY_FOR_A_WHILE);
330	ScheduleMessageService.this.updateOffset(this.delayLevel, nextOffset);	×
331	return;	×
332	} finally {
333
334	bufferCQ.release();	×
335	}
336	} // end of if (bufferCQ != null)
337	else {
338	/*
339
340
341	*/
342	long cqMinOffset = cq.getMinOffsetInQueue();	3✔
343	if (offset < cqMinOffset) {	3✔
344	failScheduleOffset = cqMinOffset;	×
345	log.error("schedule CQ offset invalid. offset=" + offset + ", cqMinOffset="	×
346	+ cqMinOffset + ", queueId=" + cq.getQueueId());	×
347	}
348	}
349	} // end of if (cq != null)
350
351	ScheduleMessageService.this.timer.schedule(new DeliverDelayedMessageTimerTask(this.delayLevel,	3✔
352	failScheduleOffset), DELAY_FOR_A_WHILE);
353	}	3✔
354
355	private MessageExtBrokerInner messageTimeup(MessageExt msgExt) {
356	MessageExtBrokerInner msgInner = new MessageExtBrokerInner();	×
357	msgInner.setBody(msgExt.getBody());	×
358	msgInner.setFlag(msgExt.getFlag());	×
359	MessageAccessor.setProperties(msgInner, msgExt.getProperties());	×
360
361	TopicFilterType topicFilterType = MessageExt.parseTopicFilterType(msgInner.getSysFlag());	×
362	long tagsCodeValue =	×
363	MessageExtBrokerInner.tagsString2tagsCode(topicFilterType, msgInner.getTags());	×
364	msgInner.setTagsCode(tagsCodeValue);	×
365	msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgExt.getProperties()));	×
366
367	msgInner.setSysFlag(msgExt.getSysFlag());	×
368	msgInner.setBornTimestamp(msgExt.getBornTimestamp());	×
369	msgInner.setBornHost(msgExt.getBornHost());	×
370	msgInner.setStoreHost(msgExt.getStoreHost());	×
371	msgInner.setReconsumeTimes(msgExt.getReconsumeTimes());	×
372
373	msgInner.setWaitStoreMsgOK(false);	×
374	MessageAccessor.clearProperty(msgInner, MessageConst.PROPERTY_DELAY_TIME_LEVEL);	×
375
376	msgInner.setTopic(msgInner.getProperty(MessageConst.PROPERTY_REAL_TOPIC));	×
377
378	String queueIdStr = msgInner.getProperty(MessageConst.PROPERTY_REAL_QUEUE_ID);	×
379	int queueId = Integer.parseInt(queueIdStr);	×
380	msgInner.setQueueId(queueId);	×
381
382	return msgInner;	×
383	}
384	}
385	}

apache / incubator-rocketmq / 654

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous