Skip to content

Commit 92e09e2

Browse files
clebertsuconicclebertsuconic
committed
ARTEMIS-5694 Closing session after a configured timeout
1 parent e407e35 commit 92e09e2

12 files changed

Lines changed: 405 additions & 62 deletions

File tree

‎artemis-commons/src/main/java/org/apache/activemq/artemis/utils/critical/CriticalComponentImpl.java‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -55,6 +55,19 @@ public CriticalCloseable measureCritical(int path) {
5555
}
5656
}
5757

58+
protected void enterCritical(int path) {
59+
if (analyzer.isMeasuring()) {
60+
measures[path].enterCritical();
61+
}
62+
}
63+
64+
protected void leaveCritical(int path) {
65+
if (analyzer.isMeasuring()) {
66+
measures[path].leaveCritical();
67+
}
68+
}
69+
70+
5871
@Override
5972
public boolean checkExpiration(long timeout, boolean reset) {
6073
for (int i = 0; i < measures.length; i++) {
@@ -64,4 +77,5 @@ public boolean checkExpiration(long timeout, boolean reset) {
6477
}
6578
return false;
6679
}
80+
6781
}

‎artemis-journal/src/main/java/org/apache/activemq/artemis/core/journal/impl/JournalImpl.java‎

Lines changed: 19 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -919,7 +919,6 @@ public void appendAddRecord(final long id,
919919
final boolean sync,
920920
final IOCompletion callback) throws Exception {
921921
checkJournalIsLoaded();
922-
lineUpContext(callback);
923922

924923
if (logger.isTraceEnabled()) {
925924
logger.trace("scheduling appendAddRecord::id={}, userRecordType={}, record = {}", id, recordType, record);
@@ -930,6 +929,8 @@ public void appendAddRecord(final long id,
930929

931930
checkRecordSize(addRecordEncodeSize, record);
932931

932+
lineUpContext(callback);
933+
933934
final SimpleFuture<Boolean> result = newSyncAndCallbackResult(sync, callback);
934935
appendExecutor.execute(() -> {
935936
journalLock.readLock().lock();
@@ -966,7 +967,6 @@ public void appendAddEvent(final long id,
966967
final boolean sync,
967968
final IOCompletion callback) throws Exception {
968969
checkJournalIsLoaded();
969-
lineUpContext(callback);
970970

971971
if (logger.isTraceEnabled()) {
972972
logger.trace("scheduling appendAddEvent::id={}, userRecordType={}, record = {}", id, recordType, record);
@@ -976,6 +976,8 @@ public void appendAddEvent(final long id,
976976

977977
checkRecordSize(addRecord.getEncodeSize(), record);
978978

979+
lineUpContext(callback);
980+
979981
final SimpleFuture<Boolean> result = newSyncAndCallbackResult(sync, callback);
980982
appendExecutor.execute(() -> {
981983
journalLock.readLock().lock();
@@ -1022,7 +1024,6 @@ public void appendUpdateRecord(final long id,
10221024
final boolean sync,
10231025
final IOCompletion callback) throws Exception {
10241026
checkJournalIsLoaded();
1025-
lineUpContext(callback);
10261027

10271028
if (logger.isTraceEnabled()) {
10281029
logger.trace("scheduling appendUpdateRecord::id={}, userRecordType={}", id, recordType);
@@ -1036,6 +1037,8 @@ public void appendUpdateRecord(final long id,
10361037
onFoundAddInfo = new SimpleFutureImpl<>();
10371038
}
10381039

1040+
lineUpContext(callback);
1041+
10391042
if (onFoundAddInfo == null) {
10401043
internalAppendUpdateRecord(id, recordType, persister, record, false, false, null, callback);
10411044
} else {
@@ -1142,7 +1145,6 @@ public void appendDeleteRecord(final long id, final boolean sync, final IOComple
11421145
}
11431146

11441147
checkJournalIsLoaded();
1145-
lineUpContext(callback);
11461148

11471149
final SimpleFuture<Boolean> onFoundAddInfo;
11481150

@@ -1152,6 +1154,8 @@ public void appendDeleteRecord(final long id, final boolean sync, final IOComple
11521154
onFoundAddInfo = new SimpleFutureImpl<>();
11531155
}
11541156

1157+
lineUpContext(callback);
1158+
11551159
if (onFoundAddInfo == null) {
11561160
internalAppendDeleteRecord(id, false, null, callback);
11571161
} else {
@@ -1448,13 +1452,13 @@ public void lineUpContext(IOCompletion callback) {
14481452
}
14491453
}
14501454

1451-
private void setErrorCondition(IOCallback otherCallback, JournalTransaction jt, Throwable t) {
1455+
private void setErrorCondition(IOCallback ioCallback, JournalTransaction jt, Throwable t) {
14521456
if (jt != null) {
14531457
jt.onError(ActiveMQExceptionType.IO_ERROR.getCode(), t.getMessage());
14541458
}
14551459

1456-
if (otherCallback != null) {
1457-
otherCallback.onError(ActiveMQExceptionType.IO_ERROR.getCode(), t.getMessage());
1460+
if (ioCallback != null) {
1461+
ioCallback.onError(ActiveMQExceptionType.IO_ERROR.getCode(), t.getMessage());
14581462
}
14591463
}
14601464

@@ -1467,9 +1471,6 @@ public void appendCommitRecord(final long txID,
14671471
final IOCompletion callback,
14681472
final boolean lineUpContext) throws Exception {
14691473
checkJournalIsLoaded();
1470-
if (lineUpContext) {
1471-
lineUpContext(callback);
1472-
}
14731474

14741475
if (logger.isTraceEnabled()) {
14751476
logger.trace("scheduling appendCommitRecord::txID={}", txID);
@@ -1480,6 +1481,9 @@ public void appendCommitRecord(final long txID,
14801481
txcheck.checkErrorCondition();
14811482
}
14821483

1484+
if (lineUpContext) {
1485+
lineUpContext(callback);
1486+
}
14831487

14841488
final SimpleFuture<JournalTransaction> result = newSyncAndCallbackResult(sync, callback);
14851489

@@ -3523,4 +3527,9 @@ public void testCompact() {
35233527
public int getCompactCount() {
35243528
return compactCount;
35253529
}
3530+
3531+
public void markTXError(long txID, Throwable t) {
3532+
JournalTransaction tx = transactions.get(txID);
3533+
tx.onError(-1, t.getMessage());
3534+
}
35263535
}

‎artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/StorageManager.java‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -117,8 +117,6 @@ default SequentialFileFactory getJournalSequentialFileFactory() {
117117
*/
118118
OperationContext getContext();
119119

120-
void lineUpContext();
121-
122120
/**
123121
* It just creates an OperationContext without associating it
124122
*/

‎artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java‎

Lines changed: 0 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -1763,16 +1763,6 @@ private void insertPersistedKeyValuePair(final PersistedKeyValuePair keyValuePai
17631763
persistedKeyValuePairs.put(keyValuePair.getKey(), keyValuePair);
17641764
}
17651765

1766-
@Override
1767-
public void lineUpContext() {
1768-
try (ArtemisCloseable lock = closeableReadLock()) {
1769-
messageJournal.lineUpContext(getContext());
1770-
}
1771-
}
1772-
1773-
// ActiveMQComponent implementation
1774-
// ------------------------------------------------------
1775-
17761766
protected abstract void beforeStart() throws Exception;
17771767

17781768
@Override

‎artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/nullpm/NullStorageManager.java‎

Lines changed: 0 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -647,10 +647,6 @@ public long storePageCounterInc(final long queueID, final int add, final long si
647647
public void commit(final long txID, final boolean lineUpContext) throws Exception {
648648
}
649649

650-
@Override
651-
public void lineUpContext() {
652-
}
653-
654650
@Override
655651
public void deletePendingLargeMessage(final long recordID) throws Exception {
656652
}

‎artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java‎

Lines changed: 60 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -108,6 +108,7 @@
108108
import org.apache.activemq.artemis.utils.JsonLoader;
109109
import org.apache.activemq.artemis.utils.PrefixUtil;
110110
import org.apache.activemq.artemis.utils.collections.TypedProperties;
111+
import org.apache.activemq.artemis.utils.critical.CriticalComponentImpl;
111112
import org.apache.activemq.artemis.utils.runnables.AtomicRunnable;
112113
import org.apache.activemq.artemis.utils.runnables.RunnableList;
113114
import org.slf4j.Logger;
@@ -116,10 +117,12 @@
116117
/**
117118
* Server side Session implementation
118119
*/
119-
public class ServerSessionImpl implements ServerSession, FailureListener {
120+
public class ServerSessionImpl extends CriticalComponentImpl implements ServerSession, FailureListener {
120121

121122
private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
122123

124+
private static final int CRITICAL_PATH_CLOSE = 0;
125+
123126
private boolean securityEnabled = true;
124127

125128
private final String securityDomain;
@@ -202,6 +205,9 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
202205
// server. Both the request and failure listener will
203206
// try to close one session from different threads
204207
// concurrently.
208+
private volatile boolean closing = false;
209+
210+
// When the doClose is called, we will make it actually closed
205211
private volatile boolean closed = false;
206212

207213
private boolean prefixEnabled = false;
@@ -237,6 +243,8 @@ public ServerSessionImpl(final String name,
237243
final Map<SimpleString, RoutingType> prefixes,
238244
final String securityDomain,
239245
boolean isLegacyProducer) throws Exception {
246+
super(server.getCriticalAnalyzer(), 1);
247+
240248
this.username = username;
241249

242250
this.password = password;
@@ -339,7 +347,7 @@ public void disableSecurity() {
339347

340348
@Override
341349
public boolean isClosed() {
342-
return closed;
350+
return closing;
343351
}
344352

345353
@Override
@@ -404,6 +412,10 @@ protected void doClose(final boolean failed) throws Exception {
404412
callback.close(failed);
405413
}
406414
synchronized (this) {
415+
if (closed) {
416+
return;
417+
}
418+
closed = true;
407419
if (server.hasBrokerSessionPlugins()) {
408420
server.callBrokerSessionPlugins(plugin -> plugin.beforeCloseSession(this, failed));
409421
}
@@ -609,7 +621,7 @@ public ServerConsumer createConsumer(final long consumerID,
609621

610622
ServerConsumer consumer;
611623
synchronized (this) {
612-
if (closed) {
624+
if (closing) {
613625
throw ActiveMQMessageBundle.BUNDLE.cannotCreateConsumerOnClosedSession(queueName);
614626
}
615627
consumer = new ServerConsumerImpl(consumerID, this, (QueueBinding) binding, filter, priority, started, browseOnly, storageManager, callback, preAcknowledge, strictUpdateDeliveryCount, managementService, supportLargeMessage, credits, server);
@@ -1777,37 +1789,45 @@ public void close(final boolean failed) {
17771789
@Override
17781790
public void close(final boolean failed, final boolean force) {
17791791
synchronized (this) {
1780-
if (closed) {
1792+
if (closing) {
17811793
return;
17821794
}
1783-
closed = true;
1795+
closing = true;
17841796
}
17851797

17861798
if (force) {
17871799
context.reset();
17881800
}
17891801

1802+
1803+
enterCritical(CRITICAL_PATH_CLOSE);
1804+
1805+
getCriticalAnalyzer().add(this);
1806+
17901807
context.executeOnCompletion(new IOCallback() {
17911808
@Override
17921809
public void onError(int errorCode, String errorMessage) {
1793-
callDoClose();
1810+
finishClose(failed);
17941811
}
17951812

17961813
@Override
17971814
public void done() {
1798-
callDoClose();
1799-
}
1800-
1801-
private void callDoClose() {
1802-
try {
1803-
doClose(failed);
1804-
} catch (Exception e) {
1805-
ActiveMQServerLogger.LOGGER.errorClosingSession(e);
1806-
}
1815+
finishClose(failed);
18071816
}
18081817
});
18091818
}
18101819

1820+
private void finishClose(boolean failed) {
1821+
try {
1822+
doClose(failed);
1823+
} catch (Exception e) {
1824+
ActiveMQServerLogger.LOGGER.errorClosingSession(e);
1825+
} finally {
1826+
leaveCritical(CRITICAL_PATH_CLOSE);
1827+
getCriticalAnalyzer().remove(this);
1828+
}
1829+
}
1830+
18111831
@Override
18121832
public void closeConsumer(final long consumerID) throws Exception {
18131833
final ServerConsumer consumer = locateConsumer(consumerID);
@@ -2192,6 +2212,30 @@ public AddressInfo getAddress(SimpleString address) {
21922212
@Override
21932213
public String toString() {
21942214
StringBuilder sb = new StringBuilder();
2215+
try {
2216+
long txID;
2217+
if (tx != null) {
2218+
txID = tx.getID();
2219+
} else {
2220+
txID = -1L;
2221+
}
2222+
sb.append("name=").append(name).append(",");
2223+
sb.append("consumers=").append(consumers.size()).append(",");
2224+
sb.append("txID=").append(txID).append(",");
2225+
sb.append("remotingConnection=").append(remotingConnection);
2226+
} catch (Throwable justLogit) {
2227+
logger.debug(justLogit.getMessage(), justLogit);
2228+
}
2229+
2230+
insertMetadata(sb);
2231+
2232+
// This will actually appear on some management operations
2233+
// so please don't clog this with debug objects
2234+
// unless you provide a special way for management to translate sessions
2235+
return "ServerSessionImpl(" + sb + ")";
2236+
}
2237+
2238+
private void insertMetadata(StringBuilder sb) {
21952239
if (this.metaData != null) {
21962240
for (Map.Entry<String, String> value : metaData.entrySet()) {
21972241
if (!sb.isEmpty()) {
@@ -2205,22 +2249,15 @@ public String toString() {
22052249
}
22062250
}
22072251
}
2208-
// This will actually appear on some management operations
2209-
// so please don't clog this with debug objects
2210-
// unless you provide a special way for management to translate sessions
2211-
return "ServerSessionImpl(" + sb.toString() + ")";
22122252
}
22132253

2214-
// FailureListener implementation
2215-
// --------------------------------------------------------------------
2216-
22172254
@Override
22182255
public void connectionFailed(final ActiveMQException me, boolean failedOver) {
22192256
/*
22202257
* This can be invoked from Netty (via channelInactive) when the connection has already been closed causing
22212258
* spurious logging about clearing up resources for failed client connections.
22222259
*/
2223-
if (closed)
2260+
if (closing)
22242261
return;
22252262

22262263
try {

‎artemis-server/src/test/java/org/apache/activemq/artemis/core/transaction/impl/TransactionImplTest.java‎

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -332,11 +332,6 @@ public OperationContext getContext() {
332332
return null;
333333
}
334334

335-
@Override
336-
public void lineUpContext() {
337-
338-
}
339-
340335
@Override
341336
public AbstractPersistedAddressSetting recoverAddressSettings(SimpleString address) {
342337
return null;

0 commit comments

Comments
 (0)