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

INTEXT-159 DefaultConnectionFactory thread safety

Browse Source 
- protect creation of connections with the lock as well, ensuring that it doesn't clash with refreshMetadata();
This commit is contained in:
Marius Bogoevici
2015-03-27 16:12:07 -04:00
committed by Artem Bilan
parent 1e6e7b9bbf
commit 790d899953
1 changed files with 27 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/core/DefaultConnectionFactory.java
View File

@@ -58,8 +58,6 @@ public class DefaultConnectionFactory implements InitializingBean, ConnectionFac
private final GetBrokersByPartitionFunction getBrokersByPartitionFunction = new GetBrokersByPartitionFunction();
private final ConnectionInstantiationFunction connectionInstantiationFunction = new ConnectionInstantiationFunction();
private final Configuration configuration;
private final AtomicReference<MetadataCache> metadataCacheHolder =
@@ -137,7 +135,33 @@ public class DefaultConnectionFactory implements InitializingBean, ConnectionFac
*/
@Override
public Connection connect(BrokerAddress brokerAddress) {
return this.kafkaBrokersCache.getIfAbsentPutWithKey(brokerAddress, connectionInstantiationFunction);
Connection connection = null;
try {
this.lock.readLock().lock();
connection = this.kafkaBrokersCache.get(brokerAddress);
}
finally {
this.lock.readLock().unlock();
}
if (connection == null) {
try {
this.lock.writeLock().lock();
connection = this.kafkaBrokersCache.get(brokerAddress);
if (connection == null) {
connection = new DefaultConnection(brokerAddress,
DefaultConnectionFactory.this.configuration.getClientId(),
DefaultConnectionFactory.this.configuration.getBufferSize(),
DefaultConnectionFactory.this.configuration.getSocketTimeout(),
DefaultConnectionFactory.this.configuration.getMinBytes(),
DefaultConnectionFactory.this.configuration.getMaxWait());
kafkaBrokersCache.put(brokerAddress, connection);
}
}
finally {
this.lock.writeLock().unlock();
}
}
return connection;
}
/**
@@ -223,21 +247,6 @@ public class DefaultConnectionFactory implements InitializingBean, ConnectionFac
}
}
@SuppressWarnings("serial")
private class ConnectionInstantiationFunction implements Function<BrokerAddress, Connection> {
@Override
public Connection valueOf(BrokerAddress brokerAddress) {
return new DefaultConnection(brokerAddress,
DefaultConnectionFactory.this.configuration.getClientId(),
DefaultConnectionFactory.this.configuration.getBufferSize(),
DefaultConnectionFactory.this.configuration.getSocketTimeout(),
DefaultConnectionFactory.this.configuration.getMinBytes(),
DefaultConnectionFactory.this.configuration.getMaxWait());
}
}
@SuppressWarnings("serial")
private class GetBrokersByPartitionFunction implements Function<Partition, BrokerAddress> {