Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #815 from artembilan/INT-2993

Browse Source 
* artembilan-INT-2993:
  INT-2993: JCMStore: fix 'idCache' race condition
This commit is contained in:
Gunnar Hillert
2013-06-06 09:55:27 -04:00
parent 22ed24c941 a76921d70b
commit 91afaedbe2
3 changed files with 115 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcChannelMessageStore.java
View File

@@ -18,12 +18,16 @@ import java.sql.SQLException;
import java.sql.Types;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import javax.sql.DataSource;
@@ -89,13 +93,21 @@ import org.springframework.util.StringUtils;
* </p
* >
* @author Gunnar Hillert
* @author Artem Bilan
* @since 2.2
*/
@ManagedResource
public class JdbcChannelMessageStore extends AbstractMessageGroupStore implements InitializingBean {
private static final Log logger = LogFactory.getLog(JdbcChannelMessageStore.class);
private final Set<String> idCache = Collections.newSetFromMap(new ConcurrentHashMap<String,Boolean>());
private final Set<String> idCache = new HashSet<String>();
private final ReadWriteLock idCacheLock = new ReentrantReadWriteLock();
private final Lock idCacheReadLock = idCacheLock.readLock();
private final Lock idCacheWriteLock = idCacheLock.writeLock();
/**
* Default value for the table prefix property.
@@ -444,16 +456,22 @@ public class JdbcChannelMessageStore extends AbstractMessageGroupStore implement
final String query;
synchronized (idCache) {
final List<Message<?>> messages;
this.idCacheReadLock.lock();
try {
if (this.usingIdCache && !this.idCache.isEmpty()) {
query = getQuery(this.channelMessageStoreQueryProvider.getPollFromGroupExcludeIdsQuery());
parameters.addValue("message_ids", idCache);
} else {
query = getQuery(this.channelMessageStoreQueryProvider.getPollFromGroupQuery());
}
messages = namedParameterJdbcTemplate.query(query, parameters, messageRowMapper);
}
finally {
this.idCacheReadLock.unlock();
}
final List<Message<?>> messages = namedParameterJdbcTemplate.query(query, parameters, messageRowMapper);
Assert.isTrue(messages.size() == 0 || messages.size() == 1);
if (messages.size() > 0){
@@ -462,11 +480,16 @@ public class JdbcChannelMessageStore extends AbstractMessageGroupStore implement
final String messageId = message.getHeaders().getId().toString();
if (this.usingIdCache) {
this.idCacheWriteLock.lock();
try {
boolean added = this.idCache.add(messageId);
boolean added = this.idCache.add(messageId);
if (logger.isDebugEnabled()) {
logger.debug(String.format("Polled message with id '%s' added: '%s'.", messageId, added));
if (logger.isDebugEnabled()) {
logger.debug(String.format("Polled message with id '%s' added: '%s'.", messageId, added));
}
}
finally {
this.idCacheWriteLock.unlock();
}
}
@@ -612,7 +635,13 @@ public class JdbcChannelMessageStore extends AbstractMessageGroupStore implement
if (logger.isDebugEnabled()) {
logger.debug("Removing Message Id:" + messageId);
}
this.idCache.remove(messageId);
this.idCacheWriteLock.lock();
try {
this.idCache.remove(messageId);
}
finally {
this.idCacheWriteLock.unlock();
}
}
/**

72
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/AbstractTxTimeoutMessageStoreTests.java
View File

@@ -12,6 +12,15 @@
*/
package org.springframework.integration.jdbc.store.channel;
import static org.junit.Assert.assertTrue;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import javax.sql.DataSource;
import org.junit.Assert;
@@ -19,20 +28,22 @@ import org.junit.Assert;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.jdbc.store.JdbcChannelMessageStore;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.annotation.Isolation;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionCallbackWithoutResult;
import org.springframework.transaction.support.TransactionTemplate;
/**
*
* @author Gunnar Hillert
*
* @author Artem Bilan
*/
abstract class AbstractTxTimeoutMessageStoreTests {
@@ -89,4 +100,61 @@ abstract class AbstractTxTimeoutMessageStoreTests {
Assert.assertEquals(Integer.valueOf(0), Integer.valueOf(testService.getDuplicateMessagesCount()));
}
public void testInt2993IdCacheConcurrency() throws InterruptedException, ExecutionException {
final String groupId = "testInt2993Group";
for (int i = 0; i < 100; i++) {
this.jdbcChannelMessageStore.addMessageToGroup(groupId, new GenericMessage<String>("testInt2993Message"));
}
ExecutorService executorService = Executors.newCachedThreadPool();
CompletionService<Boolean> completionService = new ExecutorCompletionService<Boolean>(executorService);
final int concurrency = 5;
final TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager);
for (int i = 0; i < concurrency; i++) {
completionService.submit(new Callable<Boolean>() {
@Override
public Boolean call() throws Exception {
for (int i = 0; i < 100; i++) {
boolean result = transactionTemplate.execute(new TransactionCallback<Boolean>() {
@Override
public Boolean doInTransaction(TransactionStatus status) {
Message<?> message = null;
try {
message = jdbcChannelMessageStore.pollMessageFromGroup(groupId);
}
catch (Exception e) {
log.error("IdCache race condition.", e);
return false;
}
try {
Thread.sleep(10);
}
catch (InterruptedException e) {
log.error(e);
}
if (message != null) {
jdbcChannelMessageStore.removeFromIdCache(message.getHeaders().getId().toString());
}
return true;
}
});
if (!result) return false;
}
return true;
}
});
}
for (int j = 0; j < concurrency; j++) {
assertTrue(completionService.take().get());
}
executorService.shutdown();
assertTrue(executorService.awaitTermination(5, TimeUnit.SECONDS));
}
}

10
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/store/channel/HsqlTxTimeoutMessageStoreTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2013 the original author or authors.
*
* Licensed 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
@@ -12,6 +12,8 @@
*/
package org.springframework.integration.jdbc.store.channel;
import java.util.concurrent.ExecutionException;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.test.context.ContextConfiguration;
@@ -29,9 +31,13 @@ public class HsqlTxTimeoutMessageStoreTests extends AbstractTxTimeoutMessageStor
@Test
@Override
public void test() throws InterruptedException {
super.test();
}
@Test
@Override
public void testInt2993IdCacheConcurrency() throws InterruptedException, ExecutionException {
super.testInt2993IdCacheConcurrency();
}
}