Skip to content

Commit

Permalink
Remove diamond operator for client module with JDK 1.6
Browse files Browse the repository at this point in the history
  • Loading branch information
@dongeforever
dongeforever committed May 27, 2017
1 parent e57f9ac commit 8d78175
Show file tree
Hide file tree
Showing 4 changed files with 22 additions and 21 deletions.
8 changes: 4 additions & 4 deletions ...ava/org/apache/rocketmq/client/consumer/rebalance/AllocateMessageQueueConsistentHash.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -76,19 +76,19 @@ public List<MessageQueue> allocate(String consumerGroup, String currentCID, List
}


Collection<ClientNode> cidNodes = new ArrayList<>();
Collection<ClientNode> cidNodes = new ArrayList<ClientNode>();
for (String cid : cidAll) {
cidNodes.add(new ClientNode(cid));
}

final ConsistentHashRouter<ClientNode> router; //for building hash ring
if (customHashFunction != null) {
router = new ConsistentHashRouter<>(cidNodes, virtualNodeCnt, customHashFunction);
router = new ConsistentHashRouter<ClientNode>(cidNodes, virtualNodeCnt, customHashFunction);
} else {
router = new ConsistentHashRouter<>(cidNodes, virtualNodeCnt);
router = new ConsistentHashRouter<ClientNode>(cidNodes, virtualNodeCnt);
}

List<MessageQueue> results = new ArrayList<>();
List<MessageQueue> results = new ArrayList<MessageQueue>();
for (MessageQueue mq : mqAll) {
ClientNode clientNode = router.routeNode(mq.toString());
if (clientNode != null && currentCID.equals(clientNode.getKey())) {
Expand Down
27 changes: 14 additions & 13 deletions .../org/apache/rocketmq/client/consumer/rebalance/AllocateMessageQueueConsitentHashTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
import java.util.Random;
import java.util.TreeMap;
import org.apache.rocketmq.client.consumer.AllocateMessageQueueStrategy;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.junit.Assert;
import org.junit.Before;
Expand Down Expand Up @@ -113,13 +114,13 @@ public void testAllocate(int queueSize, int consumerSize) {
//System.out.println("mqAll:" + mqAll.toString());

List<String> cidAll = createConsumerIdList(consumerSize);
List<MessageQueue> allocatedResAll = new ArrayList<>();
List<MessageQueue> allocatedResAll = new ArrayList<MessageQueue>();

Map<MessageQueue, String> allocateToAllOrigin = new TreeMap<>();
Map<MessageQueue, String> allocateToAllOrigin = new TreeMap<MessageQueue, String>();
//test allocate all
{

List<String> cidBegin = new ArrayList<>(cidAll);
List<String> cidBegin = new ArrayList<String>(cidAll);

//System.out.println("cidAll:" + cidBegin.toString());
for (String cid : cidBegin) {
Expand All @@ -135,13 +136,13 @@ public void testAllocate(int queueSize, int consumerSize) {
verifyAllocateAll(cidBegin,mqAll, allocatedResAll));
}

Map<MessageQueue, String> allocateToAllAfterRemoveOne = new TreeMap<>();
List<String> cidAfterRemoveOne = new ArrayList<>(cidAll);
Map<MessageQueue, String> allocateToAllAfterRemoveOne = new TreeMap<MessageQueue, String>();
List<String> cidAfterRemoveOne = new ArrayList<String>(cidAll);
//test allocate remove one cid
{
String removeCID = cidAfterRemoveOne.remove(0);
//System.out.println("removing one cid "+removeCID);
List<MessageQueue> mqShouldOnlyChanged = new ArrayList<>();
List<MessageQueue> mqShouldOnlyChanged = new ArrayList<MessageQueue>();
Iterator<Map.Entry<MessageQueue, String>> it = allocateToAllOrigin.entrySet().iterator();
while (it.hasNext()) {
Map.Entry<MessageQueue, String> entry = it.next();
Expand All @@ -151,7 +152,7 @@ public void testAllocate(int queueSize, int consumerSize) {
}

//System.out.println("cidAll:" + cidAfterRemoveOne.toString());
List<MessageQueue> allocatedResAllAfterRemove = new ArrayList<>();
List<MessageQueue> allocatedResAllAfterRemove = new ArrayList<MessageQueue>();
for (String cid : cidAfterRemoveOne) {
List<MessageQueue> rs = allocateMessageQueueConsistentHash.allocate("testConsumerGroup", cid, mqAll, cidAfterRemoveOne);
allocatedResAllAfterRemove.addAll(rs);
Expand All @@ -166,16 +167,16 @@ public void testAllocate(int queueSize, int consumerSize) {
verifyAfterRemove(allocateToAllOrigin, allocateToAllAfterRemoveOne, removeCID);
}

List<String> cidAfterAdd = new ArrayList<>(cidAfterRemoveOne);
List<String> cidAfterAdd = new ArrayList<String>(cidAfterRemoveOne);
//test allocate add one more cid
{
String newCid = CID_PREFIX+"NEW";
//System.out.println("add one more cid "+newCid);
cidAfterAdd.add(newCid);
List<MessageQueue> mqShouldOnlyChanged = new ArrayList<>();
List<MessageQueue> mqShouldOnlyChanged = new ArrayList<MessageQueue>();
//System.out.println("cidAll:" + cidAfterAdd.toString());
List<MessageQueue> allocatedResAllAfterAdd = new ArrayList<>();
Map<MessageQueue, String> allocateToAll3 = new TreeMap<>();
List<MessageQueue> allocatedResAllAfterAdd = new ArrayList<MessageQueue>();
Map<MessageQueue, String> allocateToAll3 = new TreeMap<MessageQueue, String>();
for (String cid : cidAfterAdd) {
List<MessageQueue> rs = allocateMessageQueueConsistentHash.allocate("testConsumerGroup", cid, mqAll, cidAfterAdd);
allocatedResAllAfterAdd.addAll(rs);
Expand Down Expand Up @@ -225,15 +226,15 @@ private void verifyAfterAdd(Map<MessageQueue, String> allocateBefore, Map<Messag
}

private List<String> createConsumerIdList(int size) {
List<String> consumerIdList = new ArrayList<>(size);
List<String> consumerIdList = new ArrayList<String>(size);
for (int i = 0; i < size; i++) {
consumerIdList.add(CID_PREFIX + String.valueOf(i));
}
return consumerIdList;
}

private List<MessageQueue> createMessageQueueList(int size) {
List<MessageQueue> messageQueueList = new ArrayList<>(size);
List<MessageQueue> messageQueueList = new ArrayList<MessageQueue>(size);
for (int i = 0; i < size; i++) {
MessageQueue mq = new MessageQueue(topic, "brokerName", i);
messageQueueList.add(mq);
Expand Down
4 changes: 2 additions & 2 deletions common/src/main/java/org/apache/rocketmq/common/consistenthash/ConsistentHashRouter.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@
* @param <T>
*/
public class ConsistentHashRouter<T extends Node> {
private final SortedMap<Long, VirtualNode<T>> ring = new TreeMap<>();
private final SortedMap<Long, VirtualNode<T>> ring = new TreeMap<Long, VirtualNode<T>>();
private final HashFunction hashFunction;

public ConsistentHashRouter(Collection<T> pNodes, int vNodeCount) {
Expand Down Expand Up @@ -64,7 +64,7 @@ public void addNode(T pNode, int vNodeCount) {
if (vNodeCount < 0) throw new IllegalArgumentException("illegal virtual node counts :" + vNodeCount);
int existingReplicas = getExistingReplicas(pNode);
for (int i = 0; i < vNodeCount; i++) {
VirtualNode<T> vNode = new VirtualNode<>(pNode, i + existingReplicas);
VirtualNode<T> vNode = new VirtualNode<T>(pNode, i + existingReplicas);
ring.put(hashFunction.hash(vNode.getKey()), vNode);
}
}
Expand Down
4 changes: 2 additions & 2 deletions store/src/main/java/org/apache/rocketmq/store/CommitLog.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -722,7 +722,7 @@ public PutMessageResult putMessages(final MessageExtBatch messageExtBatch) {

messageExtBatch.setEncodedBuff(batchEncoder.encode(messageExtBatch));

lockForPutMessage(); //spin...
putMessageLock.lock();
try {
long beginLockTimestamp = this.defaultMessageStore.getSystemClock().now();
this.beginTimeInLock = beginLockTimestamp;
Expand Down Expand Up @@ -771,7 +771,7 @@ public PutMessageResult putMessages(final MessageExtBatch messageExtBatch) {
eclipseTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginLockTimestamp;
beginTimeInLock = 0;
} finally {
releasePutMessageLock();
putMessageLock.unlock();
}


Expand Down

0 comments on commit 8d78175

Please sign in to comment.