You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by cl...@apache.org on 2018/07/25 15:05:24 UTC
[1/3] activemq-artemis git commit: ARTEMIS-1989 Replication catch up
leaking files
Repository: activemq-artemis
Updated Branches:
refs/heads/2.6.x e17c59d12 -> 47d33f3a7
ARTEMIS-1989 Replication catch up leaking files
Related commit that broke this at https://github.com/hornetq/hornetq/commit/837694e70573069cf78d1911975bef95925b6f29
(cherry picked from commit 56be281aafdbfaea2059c0ed5bd1e065f7710124)
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/be3fb960
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/be3fb960
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/be3fb960
Branch: refs/heads/2.6.x
Commit: be3fb960ed2d22243c7878725869f24a5e74de6f
Parents: e17c59d
Author: Clebert Suconic <cl...@apache.org>
Authored: Tue Jul 24 21:54:18 2018 -0400
Committer: Clebert Suconic <cl...@apache.org>
Committed: Wed Jul 25 10:51:04 2018 -0400
----------------------------------------------------------------------
.../artemis/core/io/nio/NIOSequentialFile.java | 2 +-
.../core/io/nio/NIOSequentialFileFactory.java | 2 +-
.../core/paging/impl/PagingManagerImpl.java | 13 +-
.../core/paging/impl/PagingStoreFactoryNIO.java | 38 ++++-
.../core/replication/ReplicationEndpoint.java | 9 +-
.../artemis/core/server/ActiveMQServer.java | 2 +
.../core/server/impl/ActiveMQServerImpl.java | 3 +-
...SharedNothingReplicationFlowControlTest.java | 140 ++++++++++++++++++-
8 files changed, 194 insertions(+), 15 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/be3fb960/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/nio/NIOSequentialFile.java
----------------------------------------------------------------------
diff --git a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/nio/NIOSequentialFile.java b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/nio/NIOSequentialFile.java
index 891bd5c..55654b7 100644
--- a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/nio/NIOSequentialFile.java
+++ b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/nio/NIOSequentialFile.java
@@ -37,7 +37,7 @@ import org.apache.activemq.artemis.journal.ActiveMQJournalBundle;
import org.apache.activemq.artemis.journal.ActiveMQJournalLogger;
import org.apache.activemq.artemis.utils.Env;
-public final class NIOSequentialFile extends AbstractSequentialFile {
+public class NIOSequentialFile extends AbstractSequentialFile {
private FileChannel channel;
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/be3fb960/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/nio/NIOSequentialFileFactory.java
----------------------------------------------------------------------
diff --git a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/nio/NIOSequentialFileFactory.java b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/nio/NIOSequentialFileFactory.java
index b585b24..c142377 100644
--- a/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/nio/NIOSequentialFileFactory.java
+++ b/artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/nio/NIOSequentialFileFactory.java
@@ -29,7 +29,7 @@ import org.apache.activemq.artemis.core.io.SequentialFile;
import org.apache.activemq.artemis.utils.Env;
import org.apache.activemq.artemis.utils.critical.CriticalAnalyzer;
-public final class NIOSequentialFileFactory extends AbstractSequentialFileFactory {
+public class NIOSequentialFileFactory extends AbstractSequentialFileFactory {
private static final int DEFAULT_CAPACITY_ALIGNMENT = Env.osPageSize();
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/be3fb960/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingManagerImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingManagerImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingManagerImpl.java
index 44e8067..bca70cf 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingManagerImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingManagerImpl.java
@@ -62,7 +62,7 @@ public final class PagingManagerImpl implements PagingManager {
private final HierarchicalRepository<AddressSettings> addressSettingsRepository;
- private final PagingStoreFactory pagingStoreFactory;
+ private PagingStoreFactory pagingStoreFactory;
private final AtomicLong globalSizeBytes = new AtomicLong(0);
@@ -84,6 +84,17 @@ public final class PagingManagerImpl implements PagingManager {
// Constructors
// --------------------------------------------------------------------------------------------------------------------
+
+ // for tests.. not part of the API
+ public void replacePageStoreFactory(PagingStoreFactory factory) {
+ this.pagingStoreFactory = factory;
+ }
+
+ // for tests.. not part of the API
+ public PagingStoreFactory getPagingStoreFactory() {
+ return pagingStoreFactory;
+ }
+
public PagingManagerImpl(final PagingStoreFactory pagingSPI,
final HierarchicalRepository<AddressSettings> addressSettingsRepository,
final long maxSize) {
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/be3fb960/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreFactoryNIO.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreFactoryNIO.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreFactoryNIO.java
index b2e3d4f..aa71c0e 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreFactoryNIO.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreFactoryNIO.java
@@ -63,7 +63,7 @@ public class PagingStoreFactoryNIO implements PagingStoreFactory {
private final ExecutorFactory executorFactory;
- protected final boolean syncNonTransactional;
+ private final boolean syncNonTransactional;
private PagingManager pagingManager;
@@ -71,10 +71,38 @@ public class PagingStoreFactoryNIO implements PagingStoreFactory {
private final long syncTimeout;
- protected final StorageManager storageManager;
+ private final StorageManager storageManager;
private final IOCriticalErrorListener critialErrorListener;
+ public File getDirectory() {
+ return directory;
+ }
+
+ public ExecutorFactory getExecutorFactory() {
+ return executorFactory;
+ }
+
+ public boolean isSyncNonTransactional() {
+ return syncNonTransactional;
+ }
+
+ public PagingManager getPagingManager() {
+ return pagingManager;
+ }
+
+ public long getSyncTimeout() {
+ return syncTimeout;
+ }
+
+ public StorageManager getStorageManager() {
+ return storageManager;
+ }
+
+ public IOCriticalErrorListener getCritialErrorListener() {
+ return critialErrorListener;
+ }
+
public PagingStoreFactoryNIO(final StorageManager storageManager,
final File directory,
final long syncTimeout,
@@ -135,9 +163,7 @@ public class PagingStoreFactoryNIO implements PagingStoreFactory {
factory.createDirs();
- File fileWithID = new File(directory, guid +
- File.separatorChar +
- PagingStoreFactoryNIO.ADDRESS_FILE);
+ File fileWithID = new File(directory, guid + File.separatorChar + PagingStoreFactoryNIO.ADDRESS_FILE);
try (BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(new FileOutputStream(fileWithID)))) {
writer.write(address.toString());
@@ -197,7 +223,7 @@ public class PagingStoreFactoryNIO implements PagingStoreFactory {
}
}
- private SequentialFileFactory newFileFactory(final String directoryName) {
+ protected SequentialFileFactory newFileFactory(final String directoryName) {
return new NIOSequentialFileFactory(new File(directory, directoryName), false, critialErrorListener, 1);
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/be3fb960/artemis-server/src/main/java/org/apache/activemq/artemis/core/replication/ReplicationEndpoint.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/replication/ReplicationEndpoint.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/replication/ReplicationEndpoint.java
index 6e45a8c..15d5311 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/replication/ReplicationEndpoint.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/replication/ReplicationEndpoint.java
@@ -43,8 +43,6 @@ import org.apache.activemq.artemis.core.journal.impl.JournalFile;
import org.apache.activemq.artemis.core.paging.PagedMessage;
import org.apache.activemq.artemis.core.paging.PagingManager;
import org.apache.activemq.artemis.core.paging.impl.Page;
-import org.apache.activemq.artemis.core.paging.impl.PagingManagerImpl;
-import org.apache.activemq.artemis.core.paging.impl.PagingStoreFactoryNIO;
import org.apache.activemq.artemis.core.persistence.StorageManager;
import org.apache.activemq.artemis.core.persistence.impl.journal.AbstractJournalStorageManager.JournalContent;
import org.apache.activemq.artemis.core.persistence.impl.journal.LargeServerMessageInSync;
@@ -262,7 +260,8 @@ public final class ReplicationEndpoint implements ChannelHandler, ActiveMQCompon
journalLoadInformation[jc.typeByte] = journalsHolder.get(jc).loadSyncOnly(JournalState.SYNCING);
}
- pageManager = new PagingManagerImpl(new PagingStoreFactoryNIO(storageManager, config.getPagingLocation(), config.getJournalBufferTimeout_NIO(), server.getScheduledPool(), server.getIOExecutorFactory(), config.isJournalSyncNonTransactional(), criticalErrorListener), server.getAddressSettingsRepository());
+
+ pageManager = server.createPagingManager();
pageManager.start();
@@ -446,6 +445,10 @@ public final class ReplicationEndpoint implements ChannelHandler, ActiveMQCompon
}
if (data == null) {
+ // this means close file
+ if (channel1.isOpen()) {
+ channel1.close();
+ }
return;
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/be3fb960/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServer.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServer.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServer.java
index 78ebbb7..130ce22 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServer.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServer.java
@@ -128,6 +128,8 @@ public interface ActiveMQServer extends ServiceComponent {
PagingManager getPagingManager();
+ PagingManager createPagingManager() throws Exception;
+
ManagementService getManagementService();
ActiveMQSecurityManager getSecurityManager();
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/be3fb960/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java
index 52f88cd..281be23 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java
@@ -2131,7 +2131,8 @@ public class ActiveMQServerImpl implements ActiveMQServer {
this.queueFactory = factory;
}
- protected PagingManager createPagingManager() throws Exception {
+ @Override
+ public PagingManager createPagingManager() throws Exception {
return new PagingManagerImpl(getPagingStoreFactory(), addressSettingsRepository, configuration.getGlobalMaxSize());
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/be3fb960/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/replication/SharedNothingReplicationFlowControlTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/replication/SharedNothingReplicationFlowControlTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/replication/SharedNothingReplicationFlowControlTest.java
index 981b355..5508c94 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/replication/SharedNothingReplicationFlowControlTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/replication/SharedNothingReplicationFlowControlTest.java
@@ -18,17 +18,22 @@
package org.apache.activemq.artemis.tests.integration.replication;
import java.io.File;
+import java.io.IOException;
+import java.lang.management.ManagementFactory;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.activemq.artemis.api.core.ActiveMQException;
import org.apache.activemq.artemis.api.core.Interceptor;
import org.apache.activemq.artemis.api.core.RoutingType;
+import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.api.core.client.ClientMessage;
import org.apache.activemq.artemis.api.core.client.ClientProducer;
import org.apache.activemq.artemis.api.core.client.ClientSession;
@@ -40,20 +45,34 @@ import org.apache.activemq.artemis.core.config.Configuration;
import org.apache.activemq.artemis.core.config.ha.ReplicaPolicyConfiguration;
import org.apache.activemq.artemis.core.config.ha.ReplicatedPolicyConfiguration;
import org.apache.activemq.artemis.core.config.impl.ConfigurationImpl;
+import org.apache.activemq.artemis.core.config.impl.SecurityConfiguration;
+import org.apache.activemq.artemis.core.io.IOCriticalErrorListener;
+import org.apache.activemq.artemis.core.io.SequentialFile;
import org.apache.activemq.artemis.core.io.SequentialFileFactory;
import org.apache.activemq.artemis.core.io.mapped.MappedSequentialFileFactory;
+import org.apache.activemq.artemis.core.io.nio.NIOSequentialFile;
+import org.apache.activemq.artemis.core.io.nio.NIOSequentialFileFactory;
import org.apache.activemq.artemis.core.journal.LoaderCallback;
import org.apache.activemq.artemis.core.journal.PreparedTransactionInfo;
import org.apache.activemq.artemis.core.journal.RecordInfo;
import org.apache.activemq.artemis.core.journal.impl.JournalImpl;
+import org.apache.activemq.artemis.core.paging.PagingManager;
+import org.apache.activemq.artemis.core.paging.PagingStore;
+import org.apache.activemq.artemis.core.paging.impl.PagingManagerImpl;
+import org.apache.activemq.artemis.core.paging.impl.PagingStoreFactoryNIO;
+import org.apache.activemq.artemis.core.persistence.StorageManager;
import org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds;
import org.apache.activemq.artemis.core.protocol.core.Packet;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.core.server.ActiveMQServers;
import org.apache.activemq.artemis.core.server.JournalType;
+import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
import org.apache.activemq.artemis.junit.Wait;
import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection;
+import org.apache.activemq.artemis.spi.core.security.ActiveMQJAASSecurityManager;
+import org.apache.activemq.artemis.spi.core.security.jaas.InVMLoginModule;
import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
+import org.apache.activemq.artemis.utils.ExecutorFactory;
import org.jboss.logging.Logger;
import org.junit.Assert;
import org.junit.Before;
@@ -65,7 +84,6 @@ public class SharedNothingReplicationFlowControlTest extends ActiveMQTestBase {
ExecutorService sendMessageExecutor;
-
@Before
public void setupExecutor() {
sendMessageExecutor = Executors.newCachedThreadPool();
@@ -98,8 +116,8 @@ public class SharedNothingReplicationFlowControlTest extends ActiveMQTestBase {
ClientSession sess = csf.createSession();
sess.createQueue("flowcontrol", RoutingType.ANYCAST, "flowcontrol", true);
sess.close();
- Executor sendMessageExecutor = Executors.newCachedThreadPool();
+Executor sendMessageExecutor = Executors.newCachedThreadPool();
int i = 0;
final int j = 100;
final CountDownLatch allMessageSent = new CountDownLatch(j);
@@ -193,6 +211,124 @@ public class SharedNothingReplicationFlowControlTest extends ActiveMQTestBase {
Assert.assertEquals("Backup did not replicated all journal", j, replicationCounter.get());
}
+ @Test
+ public void testSendPages() throws Exception {
+ // start live
+ Configuration liveConfiguration = createLiveConfiguration();
+ ActiveMQServer liveServer = addServer(ActiveMQServers.newActiveMQServer(liveConfiguration));
+ liveServer.start();
+
+ Wait.waitFor(() -> liveServer.isStarted());
+
+ ServerLocator locator = ServerLocatorImpl.newLocator("tcp://localhost:61616");
+ locator.setCallTimeout(60_000L);
+ locator.setConnectionTTL(60_000L);
+
+ final ClientSessionFactory csf = locator.createSessionFactory();
+ ClientSession sess = csf.createSession();
+ sess.createQueue("flowcontrol", RoutingType.ANYCAST, "flowcontrol", true);
+
+ PagingStore store = liveServer.getPagingManager().getPageStore(SimpleString.toSimpleString("flowcontrol"));
+ store.startPaging();
+
+ ClientProducer prod = sess.createProducer("flowcontrol");
+ for (int i = 0; i < 100; i++) {
+ prod.send(sess.createMessage(true));
+
+ if (i % 10 == 0) {
+ sess.commit();
+ store.forceAnotherPage();
+ }
+ }
+
+ sess.close();
+
+ openCount.set(0);
+ closeCount.set(0);
+ // start backup
+ Configuration backupConfiguration = createBackupConfiguration().setNetworkCheckURLList(null);
+
+ ActiveMQServer backupServer = new ActiveMQServerImpl(backupConfiguration, ManagementFactory.getPlatformMBeanServer(), new ActiveMQJAASSecurityManager(InVMLoginModule.class.getName(), new SecurityConfiguration())) {
+ @Override
+ public PagingManager createPagingManager() throws Exception {
+ PagingManagerImpl manager = (PagingManagerImpl) super.createPagingManager();
+ PagingStoreFactoryNIO originalPageStore = (PagingStoreFactoryNIO) manager.getPagingStoreFactory();
+ manager.replacePageStoreFactory(new PageStoreFactoryTestable(originalPageStore));
+ return manager;
+ }
+ };
+
+ addServer(backupServer).start();
+
+ Wait.waitFor(() -> backupServer.isStarted());
+
+ Wait.waitFor(backupServer::isReplicaSync, 30000);
+
+ PageStoreFactoryTestable testablePageStoreFactory = (PageStoreFactoryTestable) ((PagingManagerImpl) backupServer.getPagingManager()).getPagingStoreFactory();
+
+ Assert.assertEquals(openCount.get(), closeCount.get());
+ }
+
+ static AtomicInteger openCount = new AtomicInteger(0);
+ static AtomicInteger closeCount = new AtomicInteger(0);
+
+ private static class PageStoreFactoryTestable extends PagingStoreFactoryNIO {
+
+ PageStoreFactoryTestable(StorageManager storageManager,
+ File directory,
+ long syncTimeout,
+ ScheduledExecutorService scheduledExecutor,
+ ExecutorFactory executorFactory,
+ boolean syncNonTransactional,
+ IOCriticalErrorListener critialErrorListener) {
+ super(storageManager, directory, syncTimeout, scheduledExecutor, executorFactory, syncNonTransactional, critialErrorListener);
+ }
+
+ PageStoreFactoryTestable(PagingStoreFactoryNIO other) {
+ this(other.getStorageManager(), other.getDirectory(), other.getSyncTimeout(), other.getScheduledExecutor(), other.getExecutorFactory(), other.isSyncNonTransactional(), other.getCritialErrorListener());
+ }
+
+ @Override
+ protected SequentialFileFactory newFileFactory(String directoryName) {
+ return new TestableNIOFactory(new File(getDirectory(), directoryName), false, getCritialErrorListener(), 1);
+ }
+ }
+
+ public static class TestableNIOFactory extends NIOSequentialFileFactory {
+
+ public TestableNIOFactory(File journalDir, boolean buffered, IOCriticalErrorListener listener, int maxIO) {
+ super(journalDir, buffered, listener, maxIO);
+ }
+
+ @Override
+ public SequentialFile createSequentialFile(String fileName) {
+ return new TestableSequentialFile(this, journalDir, fileName, maxIO, writeExecutor);
+ }
+ }
+
+ public static class TestableSequentialFile extends NIOSequentialFile {
+
+ public TestableSequentialFile(SequentialFileFactory factory,
+ File directory,
+ String file,
+ int maxIO,
+ Executor writerExecutor) {
+ super(factory, directory, file, maxIO, writerExecutor);
+ }
+
+ @Override
+ public void open(int maxIO, boolean useExecutor) throws IOException {
+ super.open(maxIO, useExecutor);
+ openCount.incrementAndGet();
+ }
+
+ @Override
+ public synchronized void close() throws IOException, InterruptedException, ActiveMQException {
+ super.close();
+ closeCount.incrementAndGet();
+ }
+ }
+
// Set a small call timeout and write buffer high water mark value to trigger replication flow control
private Configuration createLiveConfiguration() throws Exception {
Configuration conf = new ConfigurationImpl();
[2/3] activemq-artemis git commit: ARTEMIS-1971 Support connection
pooling in LDAPLoginModule
Posted by cl...@apache.org.
ARTEMIS-1971 Support connection pooling in LDAPLoginModule
(cherry picked from commit d54e5a786875f7756c61e33092a0cca018426adb)
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/3e458564
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/3e458564
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/3e458564
Branch: refs/heads/2.6.x
Commit: 3e458564ee012fea55870424f3c46620974a3eac
Parents: be3fb96
Author: gtully <ga...@gmail.com>
Authored: Fri Jul 6 16:48:13 2018 +0100
Committer: Clebert Suconic <cl...@apache.org>
Committed: Wed Jul 25 11:04:15 2018 -0400
----------------------------------------------------------------------
.../spi/core/security/jaas/LDAPLoginModule.java | 12 ++-
.../core/security/jaas/LDAPLoginModuleTest.java | 77 ++++++++++++++++++++
artemis-server/src/test/resources/login.config | 21 ++++++
docs/user-manual/en/security.md | 7 ++
4 files changed, 116 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/3e458564/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/security/jaas/LDAPLoginModule.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/security/jaas/LDAPLoginModule.java b/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/security/jaas/LDAPLoginModule.java
index f8d7db5..19194fa 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/security/jaas/LDAPLoginModule.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/spi/core/security/jaas/LDAPLoginModule.java
@@ -85,6 +85,8 @@ public class LDAPLoginModule implements LoginModule {
private static final String AUTHENTICATE_USER = "authenticateUser";
private static final String REFERRAL = "referral";
private static final String PASSWORD_CODEC = "passwordCodec";
+ private static final String CONNECTION_POOL = "connectionPool";
+ private static final String CONNECTION_TIMEOUT = "connectionTimeout";
protected DirContext context;
@@ -128,7 +130,9 @@ public class LDAPLoginModule implements LoginModule {
new LDAPLoginProperty(PASSWORD_CODEC, (String) options.get(PASSWORD_CODEC)),
new LDAPLoginProperty(SASL_LOGIN_CONFIG_SCOPE, (String) options.get(SASL_LOGIN_CONFIG_SCOPE)),
new LDAPLoginProperty(AUTHENTICATE_USER, (String) options.get(AUTHENTICATE_USER)),
- new LDAPLoginProperty(REFERRAL, (String) options.get(REFERRAL))};
+ new LDAPLoginProperty(REFERRAL, (String) options.get(REFERRAL)),
+ new LDAPLoginProperty(CONNECTION_POOL, (String) options.get(CONNECTION_POOL)),
+ new LDAPLoginProperty(CONNECTION_TIMEOUT, (String) options.get(CONNECTION_TIMEOUT))};
if (isLoginPropertySet(AUTHENTICATE_USER)) {
authenticateUser = Boolean.valueOf(getLDAPPropertyValue(AUTHENTICATE_USER));
@@ -580,6 +584,12 @@ public class LDAPLoginModule implements LoginModule {
env.put(Context.SECURITY_PROTOCOL, getLDAPPropertyValue(CONNECTION_PROTOCOL));
env.put(Context.PROVIDER_URL, getLDAPPropertyValue(CONNECTION_URL));
env.put(Context.SECURITY_AUTHENTICATION, getLDAPPropertyValue(AUTHENTICATION));
+ if (isLoginPropertySet(CONNECTION_POOL)) {
+ env.put("com.sun.jndi.ldap.connect.pool", getLDAPPropertyValue(CONNECTION_POOL));
+ }
+ if (isLoginPropertySet(CONNECTION_TIMEOUT)) {
+ env.put("com.sun.jndi.ldap.connect.timeout", getLDAPPropertyValue(CONNECTION_TIMEOUT));
+ }
// handle LDAP referrals
// valid values are "throw", "ignore" and "follow"
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/3e458564/artemis-server/src/test/java/org/apache/activemq/artemis/core/security/jaas/LDAPLoginModuleTest.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/test/java/org/apache/activemq/artemis/core/security/jaas/LDAPLoginModuleTest.java b/artemis-server/src/test/java/org/apache/activemq/artemis/core/security/jaas/LDAPLoginModuleTest.java
index 97be299..d28bd4c 100644
--- a/artemis-server/src/test/java/org/apache/activemq/artemis/core/security/jaas/LDAPLoginModuleTest.java
+++ b/artemis-server/src/test/java/org/apache/activemq/artemis/core/security/jaas/LDAPLoginModuleTest.java
@@ -36,6 +36,10 @@ import java.lang.reflect.Modifier;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Hashtable;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
import org.apache.activemq.artemis.spi.core.security.jaas.JaasCallbackHandler;
import org.apache.activemq.artemis.spi.core.security.jaas.LDAPLoginModule;
@@ -134,6 +138,79 @@ public class LDAPLoginModuleTest extends AbstractLdapTestUnit {
assertTrue("no sessions after logout", waitForSessions(0));
}
+ @Test
+ public void testLoginPooled() throws LoginException {
+
+ LoginContext context = new LoginContext("LDAPLoginPooled", new CallbackHandler() {
+ @Override
+ public void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException {
+ for (int i = 0; i < callbacks.length; i++) {
+ if (callbacks[i] instanceof NameCallback) {
+ ((NameCallback) callbacks[i]).setName("first");
+ } else if (callbacks[i] instanceof PasswordCallback) {
+ ((PasswordCallback) callbacks[i]).setPassword("secret".toCharArray());
+ } else {
+ throw new UnsupportedCallbackException(callbacks[i]);
+ }
+ }
+ }
+ });
+ context.login();
+ context.logout();
+
+ // again
+
+ context.login();
+ context.logout();
+
+ // new context
+ context = new LoginContext("LDAPLoginPooled", new CallbackHandler() {
+ @Override
+ public void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException {
+ for (int i = 0; i < callbacks.length; i++) {
+ if (callbacks[i] instanceof NameCallback) {
+ ((NameCallback) callbacks[i]).setName("first");
+ } else if (callbacks[i] instanceof PasswordCallback) {
+ ((PasswordCallback) callbacks[i]).setPassword("secret".toCharArray());
+ } else {
+ throw new UnsupportedCallbackException(callbacks[i]);
+ }
+ }
+ }
+ });
+ context.login();
+ context.logout();
+
+ Executor pool = Executors.newCachedThreadPool();
+ for (int i = 0; i < 10; i++) {
+ ((ExecutorService) pool).execute(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ LoginContext context = new LoginContext("LDAPLoginPooled", new CallbackHandler() {
+ @Override
+ public void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException {
+ for (int i = 0; i < callbacks.length; i++) {
+ if (callbacks[i] instanceof NameCallback) {
+ ((NameCallback) callbacks[i]).setName("first");
+ } else if (callbacks[i] instanceof PasswordCallback) {
+ ((PasswordCallback) callbacks[i]).setPassword("secret".toCharArray());
+ } else {
+ throw new UnsupportedCallbackException(callbacks[i]);
+ }
+ }
+ }
+ });
+ context.login();
+ context.logout();
+ } catch (Exception ignored) {
+ }
+ }
+ });
+ }
+ assertTrue("no sessions after logout", waitForSessions(10));
+ }
+
private boolean waitForSessions(int expected) {
final long expiry = System.currentTimeMillis() + 5000;
int numSession = ldapServer.getLdapSessionManager().getSessions().length;
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/3e458564/artemis-server/src/test/resources/login.config
----------------------------------------------------------------------
diff --git a/artemis-server/src/test/resources/login.config b/artemis-server/src/test/resources/login.config
index 8e531ca..26791d9 100644
--- a/artemis-server/src/test/resources/login.config
+++ b/artemis-server/src/test/resources/login.config
@@ -49,6 +49,27 @@ LDAPLogin {
;
};
+LDAPLoginPooled {
+ org.apache.activemq.artemis.spi.core.security.jaas.LDAPLoginModule required
+ debug=true
+ initialContextFactory=com.sun.jndi.ldap.LdapCtxFactory
+ connectionURL="ldap://localhost:1024"
+ connectionUsername="uid=admin,ou=system"
+ connectionPassword=secret
+ connectionProtocol=s
+ authentication=simple
+ userBase="ou=system"
+ userSearchMatching="(uid={0})"
+ userSearchSubtree=false
+ roleBase="ou=system"
+ roleName=cn
+ roleSearchMatching="(member=uid={1},ou=system)"
+ roleSearchSubtree=false
+ connectionPool=true
+ connectionTimeout="2000"
+ ;
+};
+
UnAuthenticatedLDAPLogin {
org.apache.activemq.artemis.spi.core.security.jaas.LDAPLoginModule required
debug=true
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/3e458564/docs/user-manual/en/security.md
----------------------------------------------------------------------
diff --git a/docs/user-manual/en/security.md b/docs/user-manual/en/security.md
index dfd1a41..47fb228 100644
--- a/docs/user-manual/en/security.md
+++ b/docs/user-manual/en/security.md
@@ -587,6 +587,13 @@ system. It is implemented by
for the connection to the directory server. This option must be set explicitly
to an empty string, because it has no default value.
+- `connectionPool`. boolean, enable the ldap connection pool property
+ 'com.sun.jndi.ldap.connect.pool'. Note that the pool is [configured at the jvm level with system properties](https://docs.oracle.com/javase/jndi/tutorial/ldap/connect/config.html).
+
+
+- `connectionTimeout`. String milliseconds, that can time limit a ldap connection
+ attempt. The default is infinite.
+
- `userBase` - selects a particular subtree of the DIT to search for user
entries. The subtree is specified by a DN, which specifes the base node of
the subtree. For example, by setting this option to
[3/3] activemq-artemis git commit: ARTEMIS-1971 make LDAP pooling
test more robust
Posted by cl...@apache.org.
ARTEMIS-1971 make LDAP pooling test more robust
(cherry picked from commit fbdd6fe0adb65863bfc82202df143ba77d7812f6)
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/47d33f3a
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/47d33f3a
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/47d33f3a
Branch: refs/heads/2.6.x
Commit: 47d33f3a7c9d07283fe3adc8c2d91d3e787cea49
Parents: 3e45856
Author: Justin Bertram <jb...@apache.org>
Authored: Mon Jul 16 14:34:44 2018 -0500
Committer: Clebert Suconic <cl...@apache.org>
Committed: Wed Jul 25 11:04:28 2018 -0400
----------------------------------------------------------------------
.../core/security/jaas/LDAPLoginModuleTest.java | 114 ++++++++-----------
1 file changed, 45 insertions(+), 69 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/47d33f3a/artemis-server/src/test/java/org/apache/activemq/artemis/core/security/jaas/LDAPLoginModuleTest.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/test/java/org/apache/activemq/artemis/core/security/jaas/LDAPLoginModuleTest.java b/artemis-server/src/test/java/org/apache/activemq/artemis/core/security/jaas/LDAPLoginModuleTest.java
index d28bd4c..47bedd9 100644
--- a/artemis-server/src/test/java/org/apache/activemq/artemis/core/security/jaas/LDAPLoginModuleTest.java
+++ b/artemis-server/src/test/java/org/apache/activemq/artemis/core/security/jaas/LDAPLoginModuleTest.java
@@ -115,7 +115,7 @@ public class LDAPLoginModuleTest extends AbstractLdapTestUnit {
}
@Test
- public void testLogin() throws LoginException {
+ public void testLogin() throws Exception {
logger.info("num session: " + ldapServer.getLdapSessionManager().getSessions().length);
LoginContext context = new LoginContext("LDAPLogin", new CallbackHandler() {
@@ -135,100 +135,76 @@ public class LDAPLoginModuleTest extends AbstractLdapTestUnit {
context.login();
context.logout();
- assertTrue("no sessions after logout", waitForSessions(0));
+ assertTrue("sessions still active after logout", waitFor(() -> ldapServer.getLdapSessionManager().getSessions().length == 0));
}
@Test
- public void testLoginPooled() throws LoginException {
-
- LoginContext context = new LoginContext("LDAPLoginPooled", new CallbackHandler() {
- @Override
- public void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException {
- for (int i = 0; i < callbacks.length; i++) {
- if (callbacks[i] instanceof NameCallback) {
- ((NameCallback) callbacks[i]).setName("first");
- } else if (callbacks[i] instanceof PasswordCallback) {
- ((PasswordCallback) callbacks[i]).setPassword("secret".toCharArray());
- } else {
- throw new UnsupportedCallbackException(callbacks[i]);
- }
+ public void testLoginPooled() throws Exception {
+ CallbackHandler callbackHandler = callbacks -> {
+ for (int i = 0; i < callbacks.length; i++) {
+ if (callbacks[i] instanceof NameCallback) {
+ ((NameCallback) callbacks[i]).setName("first");
+ } else if (callbacks[i] instanceof PasswordCallback) {
+ ((PasswordCallback) callbacks[i]).setPassword("secret".toCharArray());
+ } else {
+ throw new UnsupportedCallbackException(callbacks[i]);
}
}
- });
+ };
+
+ LoginContext context = new LoginContext("LDAPLoginPooled", callbackHandler);
context.login();
context.logout();
// again
-
context.login();
context.logout();
// new context
- context = new LoginContext("LDAPLoginPooled", new CallbackHandler() {
- @Override
- public void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException {
- for (int i = 0; i < callbacks.length; i++) {
- if (callbacks[i] instanceof NameCallback) {
- ((NameCallback) callbacks[i]).setName("first");
- } else if (callbacks[i] instanceof PasswordCallback) {
- ((PasswordCallback) callbacks[i]).setPassword("secret".toCharArray());
- } else {
- throw new UnsupportedCallbackException(callbacks[i]);
- }
- }
- }
- });
+ context = new LoginContext("LDAPLoginPooled", callbackHandler);
context.login();
context.logout();
Executor pool = Executors.newCachedThreadPool();
- for (int i = 0; i < 10; i++) {
- ((ExecutorService) pool).execute(new Runnable() {
- @Override
- public void run() {
- try {
- LoginContext context = new LoginContext("LDAPLoginPooled", new CallbackHandler() {
- @Override
- public void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException {
- for (int i = 0; i < callbacks.length; i++) {
- if (callbacks[i] instanceof NameCallback) {
- ((NameCallback) callbacks[i]).setName("first");
- } else if (callbacks[i] instanceof PasswordCallback) {
- ((PasswordCallback) callbacks[i]).setPassword("secret".toCharArray());
- } else {
- throw new UnsupportedCallbackException(callbacks[i]);
- }
- }
- }
- });
- context.login();
- context.logout();
- } catch (Exception ignored) {
- }
+ for (int i = 0; i < 20; i++) {
+ pool.execute(() -> {
+ try {
+ LoginContext context1 = new LoginContext("LDAPLoginPooled", callbackHandler);
+ context1.login();
+ context1.logout();
+ } catch (Exception ignored) {
}
});
}
- assertTrue("no sessions after logout", waitForSessions(10));
+
+ /*
+ * The number of sessions here is variable due to the pool used to create the LoginContext objects and the pooling
+ * for the LDAP connections (which are managed by the JVM implementation). We really just need to confirm that
+ * there are still connections to the LDAP server open even after all the LoginContext objects are closed as that
+ * will indicate the LDAP connection pooling is working.
+ */
+ assertTrue("not enough active sessions after logout", waitFor(() -> ldapServer.getLdapSessionManager().getSessions().length >= 5));
+
+ ((ExecutorService) pool).shutdown();
+ ((ExecutorService) pool).awaitTermination(2, TimeUnit.SECONDS);
}
- private boolean waitForSessions(int expected) {
- final long expiry = System.currentTimeMillis() + 5000;
- int numSession = ldapServer.getLdapSessionManager().getSessions().length;
- while (numSession != expected && System.currentTimeMillis() < expiry) {
- try {
- TimeUnit.MILLISECONDS.sleep(100);
- } catch (InterruptedException ok) {
- break;
- }
- numSession = ldapServer.getLdapSessionManager().getSessions().length;
- logger.info("num session " + numSession);
+ public interface Condition {
+ boolean isSatisfied() throws Exception;
+ }
+ private boolean waitFor(final Condition condition) throws Exception {
+ final long expiry = System.currentTimeMillis() + 5000;
+ boolean conditionSatisified = condition.isSatisfied();
+ while (!conditionSatisified && System.currentTimeMillis() < expiry) {
+ TimeUnit.MILLISECONDS.sleep(100);
+ conditionSatisified = condition.isSatisfied();
}
- return numSession == expected;
+ return conditionSatisified;
}
@Test
- public void testUnauthenticated() throws LoginException {
+ public void testUnauthenticated() throws Exception {
LoginContext context = new LoginContext("UnAuthenticatedLDAPLogin", new CallbackHandler() {
@Override
public void handle(Callback[] callbacks) throws IOException, UnsupportedCallbackException {
@@ -250,7 +226,7 @@ public class LDAPLoginModuleTest extends AbstractLdapTestUnit {
return;
}
fail("Should have failed authenticating");
- assertTrue("no sessions after logout", waitForSessions(0));
+ assertTrue("sessions still active after logout", waitFor(() -> ldapServer.getLdapSessionManager().getSessions().length == 0));
}
@Test