4700

pending completion

Build # 4700

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

<a href="https://github.com/apache/rocketmq/commit/<a class=hub.com/apache/rocketmq/commit/129d7e2b387ca62e887b7b34972616e1845ab065">129d7e2b3<a href="https://github.com/apache/rocketmq/commit/129d7e2b387ca62e887b7b34972616e1845ab065">&quot;&gt;Revert &amp;quot;[ISSUE #2865] Batch message send bug fix (#2866)&amp;quot; (#2912)

This reverts commit </a><a class="double-link" href="https://github.com/apache/rocketmq/commit/<a class="double-link" href="https://github.com/apache/rocketmq/commit/16694485a7e111c601c0ec8c2a76a44cdd830067">16694485a</a>">16694485a</a><a href="https://github.com/apache/rocketmq/commit/129d7e2b387ca62e887b7b34972616e1845ab065">.

Run Details

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

17942 of 34687 relevant lines covered (51.73%)

0.52 hits per line

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

57.69

/client/src/main/java/org/apache/rocketmq/client/latency/LatencyFaultToleranceImpl.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.client.latency;

import java.util.Collections;
import java.util.Enumeration;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.rocketmq.client.common.ThreadLocalIndex;

public class LatencyFaultToleranceImpl implements LatencyFaultTolerance<String> {
    private final ConcurrentHashMap<String, FaultItem> faultItemTable = new ConcurrentHashMap<String, FaultItem>(16);

    private final ThreadLocalIndex whichItemWorst = new ThreadLocalIndex();

    @Override
    public void updateFaultItem(final String name, final long currentLatency, final long notAvailableDuration) {
        FaultItem old = this.faultItemTable.get(name);
        if (null == old) {
            final FaultItem faultItem = new FaultItem(name);
            faultItem.setCurrentLatency(currentLatency);
            faultItem.setStartTimestamp(System.currentTimeMillis() + notAvailableDuration);

            old = this.faultItemTable.putIfAbsent(name, faultItem);
            if (old != null) {
                old.setCurrentLatency(currentLatency);
                old.setStartTimestamp(System.currentTimeMillis() + notAvailableDuration);
            }
        } else {
            old.setCurrentLatency(currentLatency);
            old.setStartTimestamp(System.currentTimeMillis() + notAvailableDuration);
        }
    }

    @Override
    public boolean isAvailable(final String name) {
        final FaultItem faultItem = this.faultItemTable.get(name);
        if (faultItem != null) {
            return faultItem.isAvailable();
        }
        return true;
    }

    @Override
    public void remove(final String name) {
        this.faultItemTable.remove(name);
    }

    @Override
    public String pickOneAtLeast() {
        final Enumeration<FaultItem> elements = this.faultItemTable.elements();
        List<FaultItem> tmpList = new LinkedList<FaultItem>();
        while (elements.hasMoreElements()) {
            final FaultItem faultItem = elements.nextElement();
            tmpList.add(faultItem);
        }

        if (!tmpList.isEmpty()) {
            Collections.shuffle(tmpList);

            Collections.sort(tmpList);

            final int half = tmpList.size() / 2;
            if (half <= 0) {
                return tmpList.get(0).getName();
            } else {
                final int i = this.whichItemWorst.getAndIncrement() % half;
                return tmpList.get(i).getName();
            }
        }

        return null;
    }

    @Override
    public String toString() {
        return "LatencyFaultToleranceImpl{" +
            "faultItemTable=" + faultItemTable +
            ", whichItemWorst=" + whichItemWorst +
            '}';
    }

    class FaultItem implements Comparable<FaultItem> {
        private final String name;
        private volatile long currentLatency;
        private volatile long startTimestamp;

        public FaultItem(final String name) {
            this.name = name;
        }

        @Override
        public int compareTo(final FaultItem other) {
            if (this.isAvailable() != other.isAvailable()) {
                if (this.isAvailable())
                    return -1;

                if (other.isAvailable())
                    return 1;
            }

            if (this.currentLatency < other.currentLatency)
                return -1;
            else if (this.currentLatency > other.currentLatency) {
                return 1;
            }

            if (this.startTimestamp < other.startTimestamp)
                return -1;
            else if (this.startTimestamp > other.startTimestamp) {
                return 1;
            }

            return 0;
        }

        public boolean isAvailable() {
            return (System.currentTimeMillis() - startTimestamp) >= 0;
        }

        @Override
        public int hashCode() {
            int result = getName() != null ? getName().hashCode() : 0;
            result = 31 * result + (int) (getCurrentLatency() ^ (getCurrentLatency() >>> 32));
            result = 31 * result + (int) (getStartTimestamp() ^ (getStartTimestamp() >>> 32));
            return result;
        }

        @Override
        public boolean equals(final Object o) {
            if (this == o)
                return true;
            if (!(o instanceof FaultItem))
                return false;

            final FaultItem faultItem = (FaultItem) o;

            if (getCurrentLatency() != faultItem.getCurrentLatency())
                return false;
            if (getStartTimestamp() != faultItem.getStartTimestamp())
                return false;
            return getName() != null ? getName().equals(faultItem.getName()) : faultItem.getName() == null;

        }

        @Override
        public String toString() {
            return "FaultItem{" +
                "name='" + name + '\'' +
                ", currentLatency=" + currentLatency +
                ", startTimestamp=" + startTimestamp +
                '}';
        }

        public String getName() {
            return name;
        }

        public long getCurrentLatency() {
            return currentLatency;
        }

        public void setCurrentLatency(final long currentLatency) {
            this.currentLatency = currentLatency;
        }

        public long getStartTimestamp() {
            return startTimestamp;
        }

        public void setStartTimestamp(final long startTimestamp) {
            this.startTimestamp = startTimestamp;
        }

    }
}

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
18	package org.apache.rocketmq.client.latency;
19
20	import java.util.Collections;
21	import java.util.Enumeration;
22	import java.util.LinkedList;
23	import java.util.List;
24	import java.util.concurrent.ConcurrentHashMap;
25	import org.apache.rocketmq.client.common.ThreadLocalIndex;
26
27	public class LatencyFaultToleranceImpl implements LatencyFaultTolerance<String> {	1✔
28	private final ConcurrentHashMap<String, FaultItem> faultItemTable = new ConcurrentHashMap<String, FaultItem>(16);	1✔
29
30	private final ThreadLocalIndex whichItemWorst = new ThreadLocalIndex();	1✔
31
32	@Override
33	public void updateFaultItem(final String name, final long currentLatency, final long notAvailableDuration) {
34	FaultItem old = this.faultItemTable.get(name);	1✔
35	if (null == old) {	1✔
36	final FaultItem faultItem = new FaultItem(name);	1✔
37	faultItem.setCurrentLatency(currentLatency);	1✔
38	faultItem.setStartTimestamp(System.currentTimeMillis() + notAvailableDuration);	1✔
39
40	old = this.faultItemTable.putIfAbsent(name, faultItem);	1✔
41	if (old != null) {	1✔
42	old.setCurrentLatency(currentLatency);	×
43	old.setStartTimestamp(System.currentTimeMillis() + notAvailableDuration);	×
44	}
45	} else {	1✔
46	old.setCurrentLatency(currentLatency);	×
47	old.setStartTimestamp(System.currentTimeMillis() + notAvailableDuration);	×
48	}
49	}	1✔
50
51	@Override
52	public boolean isAvailable(final String name) {
53	final FaultItem faultItem = this.faultItemTable.get(name);	1✔
54	if (faultItem != null) {	1✔
55	return faultItem.isAvailable();	1✔
56	}
57	return true;	1✔
58	}
59
60	@Override
61	public void remove(final String name) {
62	this.faultItemTable.remove(name);	1✔
63	}	1✔
64
65	@Override
66	public String pickOneAtLeast() {
67	final Enumeration<FaultItem> elements = this.faultItemTable.elements();	1✔
68	List<FaultItem> tmpList = new LinkedList<FaultItem>();	1✔
69	while (elements.hasMoreElements()) {	1✔
70	final FaultItem faultItem = elements.nextElement();	1✔
71	tmpList.add(faultItem);	1✔
72	}	1✔
73
74	if (!tmpList.isEmpty()) {	1✔
75	Collections.shuffle(tmpList);	1✔
76
77	Collections.sort(tmpList);	1✔
78
79	final int half = tmpList.size() / 2;	1✔
80	if (half <= 0) {	1✔
81	return tmpList.get(0).getName();	1✔
82	} else {
83	final int i = this.whichItemWorst.getAndIncrement() % half;	1✔
84	return tmpList.get(i).getName();	1✔
85	}
86	}
87
88	return null;	×
89	}
90
91	@Override
92	public String toString() {
93	return "LatencyFaultToleranceImpl{" +	×
94	"faultItemTable=" + faultItemTable +
95	", whichItemWorst=" + whichItemWorst +
96	'}';
97	}
98
99	class FaultItem implements Comparable<FaultItem> {
100	private final String name;
101	private volatile long currentLatency;
102	private volatile long startTimestamp;
103
104	public FaultItem(final String name) {	1✔
105	this.name = name;	1✔
106	}	1✔
107
108	@Override
109	public int compareTo(final FaultItem other) {
110	if (this.isAvailable() != other.isAvailable()) {	1✔
111	if (this.isAvailable())	×
112	return -1;	×
113
114	if (other.isAvailable())	×
115	return 1;	×
116	}
117
118	if (this.currentLatency < other.currentLatency)	1✔
119	return -1;	×
120	else if (this.currentLatency > other.currentLatency) {	1✔
121	return 1;	1✔
122	}
123
124	if (this.startTimestamp < other.startTimestamp)	×
125	return -1;	×
126	else if (this.startTimestamp > other.startTimestamp) {	×
127	return 1;	×
128	}
129
130	return 0;	×
131	}
132
133	public boolean isAvailable() {
134	return (System.currentTimeMillis() - startTimestamp) >= 0;	1✔
135	}
136
137	@Override
138	public int hashCode() {
139	int result = getName() != null ? getName().hashCode() : 0;	×
140	result = 31 * result + (int) (getCurrentLatency() ^ (getCurrentLatency() >>> 32));	×
141	result = 31 * result + (int) (getStartTimestamp() ^ (getStartTimestamp() >>> 32));	×
142	return result;	×
143	}
144
145	@Override
146	public boolean equals(final Object o) {
147	if (this == o)	×
148	return true;	×
149	if (!(o instanceof FaultItem))	×
150	return false;	×
151
152	final FaultItem faultItem = (FaultItem) o;	×
153
154	if (getCurrentLatency() != faultItem.getCurrentLatency())	×
155	return false;	×
156	if (getStartTimestamp() != faultItem.getStartTimestamp())	×
157	return false;	×
158	return getName() != null ? getName().equals(faultItem.getName()) : faultItem.getName() == null;	×
159
160	}
161
162	@Override
163	public String toString() {
164	return "FaultItem{" +	×
165	"name='" + name + '\'' +
166	", currentLatency=" + currentLatency +
167	", startTimestamp=" + startTimestamp +
168	'}';
169	}
170
171	public String getName() {
172	return name;	1✔
173	}
174
175	public long getCurrentLatency() {
176	return currentLatency;	×
177	}
178
179	public void setCurrentLatency(final long currentLatency) {
180	this.currentLatency = currentLatency;	1✔
181	}	1✔
182
183	public long getStartTimestamp() {
184	return startTimestamp;	×
185	}
186
187	public void setStartTimestamp(final long startTimestamp) {
188	this.startTimestamp = startTimestamp;	1✔
189	}	1✔
190
191	}
192	}

apache / rocketmq / 4700

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