You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by pe...@apache.org on 2021/07/07 04:56:37 UTC
[pulsar] branch branch-2.8 updated (6981ba8 -> 14a731e)
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a change to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git.
from 6981ba8 [Broker/Bookie] Set -Dio.netty.tryReflectionSetAccessible=true for pulsar processes (#11138)
new c710c18 fix compaction entry read exception (#11175)
new fdaa859 Fix the dead lock when using hasMessageAvailableAsync and readNextAsync (#11183)
new ef7a46d Fix failing test. (#11186)
new 5f0b787 Fix some typos of the PersistentTopics (#11187)
new 936629c [Ledger] Fix ledger rollover scheduled task (#11116)
new 14a731e Fixed possible deadlock in the initialization of MLTransactionLog (#11194)
The 6 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 31 ++++++------
.../bookkeeper/mledger/impl/ManagedLedgerTest.java | 17 +++++++
.../pulsar/broker/admin/v2/PersistentTopics.java | 5 +-
.../pulsar/compaction/CompactedTopicImpl.java | 1 +
.../service/CurrentLedgerRolloverIfFullTest.java | 2 +-
.../broker/stats/ManagedLedgerMetricsTest.java | 3 +-
.../TokenAuthenticatedProducerConsumerTest.java | 36 ++++++++++++--
.../org/apache/pulsar/client/impl/ReaderTest.java | 46 ++++++++++++++++-
.../pulsar/compaction/CompactedTopicTest.java | 57 +++++++++++++++++++++-
.../apache/pulsar/client/impl/ConsumerImpl.java | 27 ++++++----
.../transaction/coordinator/TransactionLog.java | 5 ++
.../coordinator/impl/MLTransactionLogImpl.java | 46 ++++++++++++++---
.../impl/MLTransactionMetadataStoreProvider.java | 17 +++----
.../MLTransactionMetadataStoreTest.java | 13 ++++-
14 files changed, 252 insertions(+), 54 deletions(-)
[pulsar] 04/06: Fix some typos of the PersistentTopics (#11187)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 5f0b787b63bfea8a738cf8fb311ad51a05a8675c
Author: lipenghui <pe...@apache.org>
AuthorDate: Fri Jul 2 18:13:04 2021 +0800
Fix some typos of the PersistentTopics (#11187)
(cherry picked from commit 5855e0c87f3a3ee67e21db6957910092da348b8c)
---
.../java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java | 5 ++---
1 file changed, 2 insertions(+), 3 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java
index 32ec02d..20e3f75 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java
@@ -600,7 +600,6 @@ public class PersistentTopics extends PersistentTopicsBase {
@ApiParam(value = "Is authentication required to perform this operation")
@QueryParam("authoritative") @DefaultValue("false") boolean authoritative) {
validateTopicName(tenant, namespace, encodedTopic);
- validateTopicName(tenant, namespace, encodedTopic);
validateTopicPolicyOperation(topicName, PolicyName.MAX_UNACKED, PolicyOperation.WRITE);
preValidation(authoritative)
.thenCompose(__ -> internalSetMaxUnackedMessagesOnSubscription(null))
@@ -1767,7 +1766,7 @@ public class PersistentTopics extends PersistentTopicsBase {
.thenCompose(__ -> internalSetDeduplication(null))
.thenRun(() -> asyncResponse.resume(Response.noContent().build()))
.exceptionally(ex -> {
- handleTopicPolicyException("setDeduplication", ex, asyncResponse);
+ handleTopicPolicyException("removeDeduplication", ex, asyncResponse);
return null;
});
}
@@ -1858,7 +1857,7 @@ public class PersistentTopics extends PersistentTopicsBase {
asyncResponse.resume(Response.noContent().build());
})
.exceptionally(ex -> {
- handleTopicPolicyException("setRetention", ex, asyncResponse);
+ handleTopicPolicyException("removeRetention", ex, asyncResponse);
return null;
});
}
[pulsar] 03/06: Fix failing test. (#11186)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit ef7a46de3dcd78ebb0cf489efb141c7209bace0f
Author: Surinder Singh <su...@gmail.com>
AuthorDate: Fri Jul 2 00:35:53 2021 -0700
Fix failing test. (#11186)
The hard coded token has expired causing the test to fail.
Instead of hardcoding the token, we can generate the token dynamically and
avoid this problem in future
Co-authored-by: Surinder Singh <su...@splunk.com>
(cherry picked from commit 63ebc0adc0bdcccb6d32aaa6a9b080aae29f53cb)
---
.../TokenAuthenticatedProducerConsumerTest.java | 36 +++++++++++++++++++---
1 file changed, 32 insertions(+), 4 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TokenAuthenticatedProducerConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TokenAuthenticatedProducerConsumerTest.java
index 4046841..744327e 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TokenAuthenticatedProducerConsumerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/TokenAuthenticatedProducerConsumerTest.java
@@ -21,7 +21,16 @@ package org.apache.pulsar.client.api;
import static org.mockito.Mockito.spy;
import com.google.common.collect.Sets;
+import io.jsonwebtoken.Jwts;
+import io.jsonwebtoken.SignatureAlgorithm;
import java.net.URI;
+import java.security.KeyPair;
+import java.security.KeyPairGenerator;
+import java.security.NoSuchAlgorithmException;
+import java.security.PrivateKey;
+import java.time.Duration;
+import java.util.Base64;
+import java.util.Date;
import java.util.HashSet;
import java.util.Properties;
import java.util.Set;
@@ -47,10 +56,29 @@ import org.testng.annotations.Test;
public class TokenAuthenticatedProducerConsumerTest extends ProducerConsumerBase {
private static final Logger log = LoggerFactory.getLogger(TokenAuthenticatedProducerConsumerTest.class);
- // pre-create a public/private_key pair. Public key used for broker to verify client passed in token
- private final String TOKEN_PUBLIC_KEY = "data:;base64,MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAhHKgdY6arG7eE75bUPtznN5WjMu0sxLq7pI5Aaiw2Ijerbz33iO/Fdd2yJVuAZNDZPD/AVSaeliEh/BP+s2rN8KNuiywD+SlL1NGf2JDS5BvGT4Q8eHfDDRd/iY5zkK58wYwlke6C8fKCx10MTH9iYAJpzaaxs+Tu1RaatK+691aYSiMkYIfgbqAKmSCpK+48al/PkmENfuhzaTBPhCnEblhNvUhS5MjzBcAcGzecpEuVSxUzDtm8rU8DEQR6kkdXS1QnGHVNis/vgk8QzctkJKbtgDIaGzNUmDvTCyPZ8WLWSWJWb1oPxRZwpfXVP69ijU0Rme4/YkuHt6IEw6ANQIDAQAB";
- // admin token created based on private_key.
- private final String ADMIN_TOKEN = "eyJhbGciOiJSUzI1NiJ9.eyJzdWIiOiJhZG1pbiIsImV4cCI6MTYyNTEzNjQyMn0.DAfbUPZwQURgGvor4scO0NoqoyHkCulKZkhP7kksCWFvgx6B22iKuXGX42EFlFSRMWYYgIJXV7UZATCLCjJpn_ijrO6AWBmooib3f94OPoLDdkF3qXnqaLnvJtl8_sCoLCSghR_O3hQFgQW2GRjMDdfJgl2_HXCWuzedtI5cQJdbpfU0NU10nzo7RtrpCmUdgQYQEHegYOawLqQVvr53ZGjrZilBXY9HHz1mSlnwZGNGVNNdvRthBuGtXtfKgtfSDF5jLqABvK8TUpdNJybibeiOspdzuY19-wVt4eVXzNAGsP4V4Zs91MgIUYV5lWKnBUuVWalppkMWhRF4Jf-KWQ";
+ private final String ADMIN_TOKEN;
+ private final String TOKEN_PUBLIC_KEY;
+
+ TokenAuthenticatedProducerConsumerTest() throws NoSuchAlgorithmException {
+ KeyPairGenerator kpg = KeyPairGenerator.getInstance("RSA");
+ KeyPair kp = kpg.generateKeyPair();
+
+ byte[] encodedPublicKey = kp.getPublic().getEncoded();
+ TOKEN_PUBLIC_KEY = "data:;base64," + Base64.getEncoder().encodeToString(encodedPublicKey);
+ ADMIN_TOKEN = generateToken(kp);
+ }
+
+ private String generateToken(KeyPair kp) {
+ PrivateKey pkey = kp.getPrivate();
+ long expMillis = System.currentTimeMillis() + Duration.ofHours(1).toMillis();
+ Date exp = new Date(expMillis);
+
+ return Jwts.builder()
+ .setSubject("admin")
+ .setExpiration(exp)
+ .signWith(pkey, SignatureAlgorithm.forSigningKey(pkey))
+ .compact();
+ }
@BeforeMethod
@Override
[pulsar] 01/06: fix compaction entry read exception (#11175)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit c710c187d266f0968ad48601636c653912ad5b49
Author: hangc0276 <ch...@apache.org>
AuthorDate: Fri Jul 2 21:34:10 2021 +0800
fix compaction entry read exception (#11175)
(cherry picked from commit c716495a4282093bd81f9cd43f909811213bfb32)
---
.../pulsar/compaction/CompactedTopicImpl.java | 1 +
.../pulsar/compaction/CompactedTopicTest.java | 57 +++++++++++++++++++++-
2 files changed, 57 insertions(+), 1 deletion(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java
index e0755a7..12748e8 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java
@@ -109,6 +109,7 @@ public class CompactedTopicImpl implements CompactedTopic {
if (startPoint == NEWER_THAN_COMPACTED) {
cursor.seek(compactionHorizon.getNext());
callback.readEntriesComplete(Collections.emptyList(), ctx);
+ return CompletableFuture.completedFuture(null);
}
return readEntries(context.ledger, startPoint, endPoint)
.thenAccept((entries) -> {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicTest.java
index 2d3411b..410d1e5 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactedTopicTest.java
@@ -46,14 +46,16 @@ import org.apache.commons.lang3.tuple.Triple;
import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
import org.apache.pulsar.client.admin.LongRunningProcessStatus;
import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.ProducerBuilder;
import org.apache.pulsar.client.api.RawMessage;
+import org.apache.pulsar.client.api.Reader;
+import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.impl.RawMessageImpl;
import org.apache.pulsar.common.api.proto.MessageIdData;
import org.apache.pulsar.common.policies.data.ClusterData;
-import org.apache.pulsar.common.policies.data.ClusterDataImpl;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
import org.apache.pulsar.common.util.FutureUtil;
import org.testng.Assert;
@@ -305,4 +307,57 @@ public class CompactedTopicTest extends MockedPulsarServiceBaseTest {
producer.close();
}
+
+ @Test(timeOut = 30000)
+ public void testReadMessageFromCompactedLedger() throws Exception {
+ final String key = "1";
+ String msg = "test compaction msg";
+ final String topic = "persistent://my-property/use/my-ns/testCompactWithEmptyMessage-" + UUID.randomUUID();
+ admin.topics().createPartitionedTopic(topic, 1);
+ final int numMessages = 10;
+
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING).topic(topic).enableBatching(false).create();
+ for (int i = 0; i < numMessages; ++i) {
+ producer.newMessage().key(key).value(msg).send();
+ }
+
+ admin.topics().triggerCompaction(topic);
+ boolean succeed = retryStrategically((test) -> {
+ try {
+ return LongRunningProcessStatus.Status.SUCCESS.equals(admin.topics().compactionStatus(topic).status);
+ } catch (PulsarAdminException e) {
+ return false;
+ }
+ }, 10, 200);
+
+ Assert.assertTrue(succeed);
+
+
+ final String newKey = "2";
+ String newMsg = "test compaction msg v2";
+ for (int i = 0; i < numMessages; ++i) {
+ producer.newMessage().key(newKey).value(newMsg).send();
+ }
+
+ Reader<String> reader = pulsarClient.newReader(Schema.STRING)
+ .topic(topic)
+ .subscriptionName("test")
+ .readCompacted(true)
+ .startMessageId(MessageId.earliest)
+ .create();
+
+ int compactedMsgCount = 0;
+ int nonCompactedMsgCount = 0;
+ while (reader.hasMessageAvailable()) {
+ Message<String> message = reader.readNext();
+ if (key.equals(message.getKey()) && msg.equals(message.getValue())) {
+ compactedMsgCount++;
+ } else if (newKey.equals(message.getKey()) && newMsg.equals(message.getValue())) {
+ nonCompactedMsgCount++;
+ }
+ }
+
+ Assert.assertEquals(compactedMsgCount, 1);
+ Assert.assertEquals(nonCompactedMsgCount, numMessages);
+ }
}
[pulsar] 02/06: Fix the dead lock when using
hasMessageAvailableAsync and readNextAsync (#11183)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit fdaa85945be16e3a6ac3088f3cec8e4d488db199
Author: lipenghui <pe...@apache.org>
AuthorDate: Fri Jul 2 19:25:43 2021 +0800
Fix the dead lock when using hasMessageAvailableAsync and readNextAsync (#11183)
The issue will happens after satisfying the following conditions:
1. The messages are added to the incoming queue before reading messages
2. The result future of the readNextAsync been complete before call future.whenComplete by users,
This won't always appear.
After that, since we are using the IO thread to call the callback of the hasMessageAvailableAsync,
so the IO thread will process the message.getValue(). Then might get a deadlock as followings:
```
java.util.concurrent.CompletableFuture.get() CompletableFuture.java:1998
org.apache.pulsar.client.impl.schema.reader.AbstractMultiVersionReader.getSchemaInfoByVersion(byte[]) AbstractMultiVersionReader.java:115
org.apache.pulsar.client.impl.schema.reader.MultiVersionAvroReader.loadReader(BytesSchemaVersion) MultiVersionAvroReader.java:47
org.apache.pulsar.client.impl.schema.reader.AbstractMultiVersionReader$1.load(BytesSchemaVersion) AbstractMultiVersionReader.java:52
org.apache.pulsar.client.impl.schema.reader.AbstractMultiVersionReader$1.load(Object) AbstractMultiVersionReader.java:49
com.google.common.cache.LocalCache$LoadingValueReference.loadFuture(Object, CacheLoader) LocalCache.java:3529
com.google.common.cache.LocalCache$Segment.loadSync(Object, int, LocalCache$LoadingValueReference, CacheLoader) LocalCache.java:2278
com.google.common.cache.LocalCache$Segment.lockedGetOrLoad(Object, int, CacheLoader) LocalCache.java:2155
com.google.common.cache.LocalCache$Segment.get(Object, int, CacheLoader) LocalCache.java:2045
com.google.common.cache.LocalCache.get(Object, CacheLoader) LocalCache.java:3951
com.google.common.cache.LocalCache.getOrLoad(Object) LocalCache.java:3974
com.google.common.cache.LocalCache$LocalLoadingCache.get(Object) LocalCache.java:4935
org.apache.pulsar.client.impl.schema.reader.AbstractMultiVersionReader.read(byte[], byte[]) AbstractMultiVersionReader.java:86
org.apache.pulsar.client.impl.schema.AbstractStructSchema.decode(byte[], byte[]) AbstractStructSchema.java:60
org.apache.pulsar.client.impl.MessageImpl.getValue() MessageImpl.java:301
org.apache.pulsar.broker.service.SystemTopicBasedTopicPoliciesService.refreshTopicPoliciesCache(Message) SystemTopicBasedTopicPoliciesService.java:302
org.apache.pulsar.broker.service.SystemTopicBasedTopicPoliciesService.lambda$initPolicesCache$7(SystemTopicClient$Reader, Throwable, CompletableFuture, Message, Throwable) SystemTopicBasedTopicPoliciesService.java:254
org.apache.pulsar.broker.service.SystemTopicBasedTopicPoliciesService$$Lambda$817.accept(Object, Object)
java.util.concurrent.CompletableFuture.uniWhenComplete(Object, BiConsumer, CompletableFuture$UniWhenComplete) CompletableFuture.java:859
java.util.concurrent.CompletableFuture.uniWhenCompleteStage(Executor, BiConsumer) CompletableFuture.java:883
java.util.concurrent.CompletableFuture.whenComplete(BiConsumer) CompletableFuture.java:2251
org.apache.pulsar.broker.service.SystemTopicBasedTopicPoliciesService.lambda$initPolicesCache$10(SystemTopicClient$Reader, CompletableFuture, Boolean, Throwable) SystemTopicBasedTopicPoliciesService.java:246
org.apache.pulsar.broker.service.SystemTopicBasedTopicPoliciesService$$Lambda$725.accept(Object, Object) org.apache.pulsar.client.impl.ClientCnx.handleGetLastMessageIdSuccess(PulsarApi$CommandGetLastMessageIdResponse) ClientCnx.java:468
```
Since we are introduced the internal thread pool for handling the client internal executions.
So the fix is using the internal thread to process the callback of the hasMessageAvailableAsync
(cherry picked from commit ed42007d5f5df063a61626e49803bbf35c5a3eff)
---
.../org/apache/pulsar/client/impl/ReaderTest.java | 46 +++++++++++++++++++++-
.../apache/pulsar/client/impl/ConsumerImpl.java | 27 ++++++++-----
2 files changed, 62 insertions(+), 11 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ReaderTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ReaderTest.java
index a04f73b..4a2466f 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ReaderTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ReaderTest.java
@@ -32,8 +32,11 @@ import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
+
import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
@@ -51,10 +54,11 @@ import org.apache.pulsar.client.api.ReaderBuilder;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.ClusterData;
-import org.apache.pulsar.common.policies.data.ClusterDataImpl;
import org.apache.pulsar.common.policies.data.RetentionPolicies;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
import org.apache.pulsar.common.util.Murmur3_32Hash;
+import org.apache.pulsar.schema.Schemas;
+import org.awaitility.Awaitility;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
@@ -510,4 +514,44 @@ public class ReaderTest extends MockedPulsarServiceBaseTest {
}
+ @Test(timeOut = 30000)
+ public void testAvoidUsingIoThreadToGetValueOfMessage() throws Exception {
+ final String topic = "persistent://my-property/my-ns/testAvoidUsingIoThreadToGetValueOfMessage";
+
+ @Cleanup
+ Producer<Schemas.PersonOne> producer = pulsarClient.newProducer(Schema.AVRO(Schemas.PersonOne.class))
+ .topic(topic)
+ .create();
+
+ producer.send(new Schemas.PersonOne(1));
+
+ @Cleanup
+ Reader<Schemas.PersonOne> reader = pulsarClient.newReader(Schema.AVRO(Schemas.PersonOne.class))
+ .topic(topic)
+ .startMessageId(MessageId.earliest)
+ .create();
+
+ CountDownLatch latch = new CountDownLatch(1);
+ List<Schemas.PersonOne> received = new ArrayList<>(1);
+ // Make sure the message is added to the incoming queue
+ Awaitility.await().untilAsserted(() ->
+ assertTrue(((ReaderImpl<?>) reader).getConsumer().incomingMessages.size() > 0));
+ reader.hasMessageAvailableAsync().whenComplete((has, e) -> {
+ if (e == null && has) {
+ CompletableFuture<Message<Schemas.PersonOne>> future = reader.readNextAsync();
+ // Make sure the future completed
+ Awaitility.await().pollInterval(1, TimeUnit.MILLISECONDS).untilAsserted(future::isDone);
+ future.whenComplete((msg, ex) -> {
+ if (ex == null) {
+ received.add(msg.getValue());
+ }
+ latch.countDown();
+ });
+ } else {
+ latch.countDown();
+ }
+ });
+ latch.await();
+ Assert.assertEquals(received.size(), 1);
+ }
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
index c5301da..a0fb143 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
@@ -1906,14 +1906,15 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
.compare(markDeletePosition.getEntryId(), lastMessageId.getEntryId())
.result();
if (lastMessageId.getEntryId() < 0) {
- booleanFuture.complete(false);
+ completehasMessageAvailableWithValue(booleanFuture, false);
} else {
- booleanFuture.complete(resetIncludeHead ? result <= 0 : result < 0);
+ completehasMessageAvailableWithValue(booleanFuture,
+ resetIncludeHead ? result <= 0 : result < 0);
}
} else if (lastMessageId == null || lastMessageId.getEntryId() < 0) {
- booleanFuture.complete(false);
+ completehasMessageAvailableWithValue(booleanFuture, false);
} else {
- booleanFuture.complete(resetIncludeHead);
+ completehasMessageAvailableWithValue(booleanFuture, resetIncludeHead);
}
}).exceptionally(ex -> {
log.error("[{}][{}] Failed getLastMessageId command", topic, subscription, ex);
@@ -1925,16 +1926,16 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
}
if (hasMoreMessages(lastMessageIdInBroker, startMessageId, resetIncludeHead)) {
- booleanFuture.complete(true);
+ completehasMessageAvailableWithValue(booleanFuture, true);
return booleanFuture;
}
getLastMessageIdAsync().thenAccept(messageId -> {
lastMessageIdInBroker = messageId;
if (hasMoreMessages(lastMessageIdInBroker, startMessageId, resetIncludeHead)) {
- booleanFuture.complete(true);
+ completehasMessageAvailableWithValue(booleanFuture, true);
} else {
- booleanFuture.complete(false);
+ completehasMessageAvailableWithValue(booleanFuture, false);
}
}).exceptionally(e -> {
log.error("[{}][{}] Failed getLastMessageId command", topic, subscription);
@@ -1945,16 +1946,16 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
} else {
// read before, use lastDequeueMessage for comparison
if (hasMoreMessages(lastMessageIdInBroker, lastDequeuedMessageId, false)) {
- booleanFuture.complete(true);
+ completehasMessageAvailableWithValue(booleanFuture, true);
return booleanFuture;
}
getLastMessageIdAsync().thenAccept(messageId -> {
lastMessageIdInBroker = messageId;
if (hasMoreMessages(lastMessageIdInBroker, lastDequeuedMessageId, false)) {
- booleanFuture.complete(true);
+ completehasMessageAvailableWithValue(booleanFuture, true);
} else {
- booleanFuture.complete(false);
+ completehasMessageAvailableWithValue(booleanFuture, false);
}
}).exceptionally(e -> {
log.error("[{}][{}] Failed getLastMessageId command", topic, subscription);
@@ -1966,6 +1967,12 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
return booleanFuture;
}
+ private void completehasMessageAvailableWithValue(CompletableFuture<Boolean> future, boolean value) {
+ internalPinnedExecutor.execute(() -> {
+ future.complete(value);
+ });
+ }
+
private boolean hasMoreMessages(MessageId lastMessageIdInBroker, MessageId messageId, boolean inclusive) {
if (inclusive && lastMessageIdInBroker.compareTo(messageId) >= 0 &&
((MessageIdImpl) lastMessageIdInBroker).getEntryId() != -1) {
[pulsar] 05/06: [Ledger] Fix ledger rollover scheduled task (#11116)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 936629cc684b1d45ebc47d3066f67bf7dbf9a067
Author: ran <ga...@126.com>
AuthorDate: Sat Jul 3 01:40:37 2021 +0800
[Ledger] Fix ledger rollover scheduled task (#11116)
### Motivation
Currently, the ledger rollover scheduled task will execute before reach the ledger maximum rollover time, this will cause the ledger doesn't roll over in time.
### Modifications
Only make the ledger rollover scheduled task after the ledger created successfully.
If the scheduled task was executed when there is no entry in the current ledger, the scheduled task will not be re-executed, and if there is new entry is added the ledger will rollover.
### Verifying this change
Add a unit test to verify the ledger could be rolled over in time.
(cherry picked from commit a092ceb63588c94ad12327464e65b5fb69d1569e)
---
.../bookkeeper/mledger/impl/ManagedLedgerImpl.java | 31 ++++++++++++----------
.../bookkeeper/mledger/impl/ManagedLedgerTest.java | 17 ++++++++++++
.../service/CurrentLedgerRolloverIfFullTest.java | 2 +-
3 files changed, 35 insertions(+), 15 deletions(-)
diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
index 7677c71..9a5cea0 100644
--- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
+++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
@@ -401,8 +401,6 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
});
scheduleTimeoutTask();
-
- scheduleRollOverLedgerTask();
}
private synchronized void initializeBookKeeper(final ManagedLedgerInitializeLedgerCallback callback) {
@@ -467,7 +465,7 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
log.info("[{}] Created ledger {}", name, lh.getId());
STATE_UPDATER.set(this, State.LedgerOpened);
- lastLedgerCreatedTimestamp = clock.millis();
+ updateLastLedgerCreatedTimeAndScheduleRolloverTask();
currentLedger = lh;
lastConfirmedEntry = new PositionImpl(lh.getId(), -1);
@@ -1488,7 +1486,7 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
public synchronized void updateLedgersIdsComplete(Stat stat) {
STATE_UPDATER.set(this, State.LedgerOpened);
- lastLedgerCreatedTimestamp = clock.millis();
+ updateLastLedgerCreatedTimeAndScheduleRolloverTask();
if (log.isDebugEnabled()) {
log.debug("[{}] Resending {} pending messages", name, pendingAddEntries.size());
@@ -3361,7 +3359,7 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
|| currentLedgerSize >= (config.getMaxSizePerLedgerMb() * MegaByte));
long timeSinceLedgerCreationMs = clock.millis() - lastLedgerCreatedTimestamp;
- boolean maxLedgerTimeReached = timeSinceLedgerCreationMs >= maximumRolloverTimeMs;
+ boolean maxLedgerTimeReached = timeSinceLedgerCreationMs >= config.getMaximumRolloverTimeMs();
if (spaceQuotaReached || maxLedgerTimeReached) {
if (config.getMinimumRolloverTimeMs() > 0) {
@@ -3663,15 +3661,6 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
}
}
- private void scheduleRollOverLedgerTask() {
- if (config.getMaximumRolloverTimeMs() > 0) {
- long interval = config.getMaximumRolloverTimeMs();
- this.checkLedgerRollTask = this.scheduledExecutor.scheduleAtFixedRate(safeRun(() -> {
- rollCurrentLedgerIfFull();
- }), interval, interval, TimeUnit.MILLISECONDS);
- }
- }
-
private void checkAddTimeout() {
long timeoutSec = config.getAddEntryTimeoutSeconds();
if (timeoutSec < 1) {
@@ -3929,4 +3918,18 @@ public class ManagedLedgerImpl implements ManagedLedger, CreateCallback {
return CompletableFuture.completedFuture(ensembles);
});
}
+
+ private void updateLastLedgerCreatedTimeAndScheduleRolloverTask() {
+ this.lastLedgerCreatedTimestamp = clock.millis();
+ if (config.getMaximumRolloverTimeMs() > 0) {
+ if (checkLedgerRollTask != null && !checkLedgerRollTask.isDone()) {
+ // new ledger has been created successfully
+ // and the previous checkLedgerRollTask is not done, we could cancel it
+ checkLedgerRollTask.cancel(true);
+ }
+ this.checkLedgerRollTask = this.scheduledExecutor.schedule(
+ safeRun(this::rollCurrentLedgerIfFull), getMaximumRolloverTimeMs(config), TimeUnit.MILLISECONDS);
+ }
+ }
+
}
diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
index e0b15fe..d55598c 100644
--- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
+++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerTest.java
@@ -2923,6 +2923,23 @@ public class ManagedLedgerTest extends MockedBookKeeperTestCase {
}
@Test
+ public void testLedgerReachMaximumRolloverTime() throws Exception {
+ ManagedLedgerConfig config = new ManagedLedgerConfig();
+ config.setMinimumRolloverTime(1, TimeUnit.MILLISECONDS);
+ config.setMaximumRolloverTime(1, TimeUnit.SECONDS);
+
+ ManagedLedger ml = factory.open("ledger-reach-maximum-rollover-time", config);
+ long firstLedgerId = ml.addEntry("test".getBytes()).getLedgerId();
+
+ // the ledger rollover scheduled time is between 1000 and 1050 ms,
+ // wait 1100 ms, the ledger should be rolled over.
+ Awaitility.await()
+ .atMost(1100, TimeUnit.MILLISECONDS)
+ .pollInterval(100, TimeUnit.MILLISECONDS)
+ .until(() -> firstLedgerId != ml.addEntry("test".getBytes()).getLedgerId());
+ }
+
+ @Test
public void testExpiredLedgerDeletionAfterManagedLedgerRestart() throws Exception {
ManagedLedgerConfig config = new ManagedLedgerConfig();
config.setRetentionTime(1, TimeUnit.SECONDS);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/CurrentLedgerRolloverIfFullTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/CurrentLedgerRolloverIfFullTest.java
index 1bb8dcb..77ec229 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/CurrentLedgerRolloverIfFullTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/CurrentLedgerRolloverIfFullTest.java
@@ -72,7 +72,7 @@ public class CurrentLedgerRolloverIfFullTest extends BrokerTestBase {
managedLedgerConfig.setRetentionTime(1, TimeUnit.SECONDS);
managedLedgerConfig.setMaxEntriesPerLedger(2);
managedLedgerConfig.setMinimumRolloverTime(1, TimeUnit.MILLISECONDS);
- managedLedgerConfig.setMaximumRolloverTime(5, TimeUnit.MILLISECONDS);
+ managedLedgerConfig.setMaximumRolloverTime(1, TimeUnit.SECONDS);
int msgNum = 10;
for (int i = 0; i < msgNum; i++) {
[pulsar] 06/06: Fixed possible deadlock in the initialization of
MLTransactionLog (#11194)
Posted by pe...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
penghui pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 14a731e88f46f5c565acafc1b9944b4e7a8293be
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Fri Jul 2 18:56:14 2021 +0200
Fixed possible deadlock in the initialization of MLTransactionLog (#11194)
* Fixed possible deadlock in the initialization of MLTransactionLog
* Fixed tests
(cherry picked from commit 555042a45d9692148c302fbdedbb47f1f2572a48)
---
.../broker/stats/ManagedLedgerMetricsTest.java | 3 +-
.../transaction/coordinator/TransactionLog.java | 5 +++
.../coordinator/impl/MLTransactionLogImpl.java | 46 +++++++++++++++++++---
.../impl/MLTransactionMetadataStoreProvider.java | 17 +++-----
.../MLTransactionMetadataStoreTest.java | 13 +++++-
5 files changed, 64 insertions(+), 20 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/ManagedLedgerMetricsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/ManagedLedgerMetricsTest.java
index 9d19e3e..98ac261 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/ManagedLedgerMetricsTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/ManagedLedgerMetricsTest.java
@@ -103,7 +103,8 @@ public class ManagedLedgerMetricsTest extends BrokerTestBase {
ManagedLedgerConfig managedLedgerConfig = new ManagedLedgerConfig();
managedLedgerConfig.setMaxEntriesPerLedger(2);
new MLTransactionLogImpl(TransactionCoordinatorID.get(0),
- pulsar.getManagedLedgerFactory(), managedLedgerConfig);
+ pulsar.getManagedLedgerFactory(), managedLedgerConfig)
+ .initialize().join();
ManagedLedgerMetrics metrics = new ManagedLedgerMetrics(pulsar);
metrics.generate();
}
diff --git a/pulsar-transaction/coordinator/src/main/java/org/apache/pulsar/transaction/coordinator/TransactionLog.java b/pulsar-transaction/coordinator/src/main/java/org/apache/pulsar/transaction/coordinator/TransactionLog.java
index 7f70c28..a752c3b 100644
--- a/pulsar-transaction/coordinator/src/main/java/org/apache/pulsar/transaction/coordinator/TransactionLog.java
+++ b/pulsar-transaction/coordinator/src/main/java/org/apache/pulsar/transaction/coordinator/TransactionLog.java
@@ -27,6 +27,11 @@ import org.apache.pulsar.transaction.coordinator.proto.TransactionMetadataEntry;
*/
public interface TransactionLog {
+ /**
+ * Initialize the TransactionLog implementation
+ */
+ CompletableFuture<Void> initialize();
+
/**
* Replay transaction log to load the transaction map.
diff --git a/pulsar-transaction/coordinator/src/main/java/org/apache/pulsar/transaction/coordinator/impl/MLTransactionLogImpl.java b/pulsar-transaction/coordinator/src/main/java/org/apache/pulsar/transaction/coordinator/impl/MLTransactionLogImpl.java
index 4820805..4098cfd 100644
--- a/pulsar-transaction/coordinator/src/main/java/org/apache/pulsar/transaction/coordinator/impl/MLTransactionLogImpl.java
+++ b/pulsar-transaction/coordinator/src/main/java/org/apache/pulsar/transaction/coordinator/impl/MLTransactionLogImpl.java
@@ -52,11 +52,13 @@ public class MLTransactionLogImpl implements TransactionLog {
private static final Logger log = LoggerFactory.getLogger(MLTransactionLogImpl.class);
- private final ManagedLedger managedLedger;
+ private final ManagedLedgerFactory managedLedgerFactory;
+ private final ManagedLedgerConfig managedLedgerConfig;
+ private ManagedLedger managedLedger;
public final static String TRANSACTION_LOG_PREFIX = "__transaction_log_";
- private final ManagedCursor cursor;
+ private ManagedCursor cursor;
public final static String TRANSACTION_SUBSCRIPTION_NAME = "transaction.subscription";
@@ -70,19 +72,51 @@ public class MLTransactionLogImpl implements TransactionLog {
public MLTransactionLogImpl(TransactionCoordinatorID tcID,
ManagedLedgerFactory managedLedgerFactory,
- ManagedLedgerConfig managedLedgerConfig) throws Exception {
+ ManagedLedgerConfig managedLedgerConfig) {
this.topicName = TopicName.get(TopicDomain.persistent.value(),
NamespaceName.SYSTEM_NAMESPACE, TRANSACTION_LOG_PREFIX + tcID.getId());
this.tcId = tcID.getId();
this.mlTransactionLogInterceptor = new MLTransactionLogInterceptor();
managedLedgerConfig.setManagedLedgerInterceptor(this.mlTransactionLogInterceptor);
- this.managedLedger = managedLedgerFactory.open(topicName.getPersistenceNamingEncoding(), managedLedgerConfig);
- this.cursor = managedLedger.openCursor(TRANSACTION_SUBSCRIPTION_NAME,
- CommandSubscribe.InitialPosition.Earliest);
+ this.managedLedgerFactory = managedLedgerFactory;
+ this.managedLedgerConfig = managedLedgerConfig;
this.entryQueue = new SpscArrayQueue<>(2000);
}
@Override
+ public CompletableFuture<Void> initialize() {
+ CompletableFuture<Void> future = new CompletableFuture<>();
+ managedLedgerFactory.asyncOpen(topicName.getPersistenceNamingEncoding(),
+ managedLedgerConfig,
+ new AsyncCallbacks.OpenLedgerCallback() {
+ @Override
+ public void openLedgerComplete(ManagedLedger ledger, Object ctx) {
+ MLTransactionLogImpl.this.managedLedger = ledger;
+
+ managedLedger.asyncOpenCursor(TRANSACTION_SUBSCRIPTION_NAME,
+ CommandSubscribe.InitialPosition.Earliest, new AsyncCallbacks.OpenCursorCallback() {
+ @Override
+ public void openCursorComplete(ManagedCursor cursor, Object ctx) {
+ MLTransactionLogImpl.this.cursor = cursor;
+ future.complete(null);
+ }
+
+ @Override
+ public void openCursorFailed(ManagedLedgerException exception, Object ctx) {
+ future.completeExceptionally(exception);
+ }
+ }, null);
+ }
+
+ @Override
+ public void openLedgerFailed(ManagedLedgerException exception, Object ctx) {
+ future.completeExceptionally(exception);
+ }
+ }, null, null);
+ return future;
+ }
+
+ @Override
public void replayAsync(TransactionLogReplayCallback transactionLogReplayCallback) {
new TransactionLogReplayer(transactionLogReplayCallback).start();
}
diff --git a/pulsar-transaction/coordinator/src/main/java/org/apache/pulsar/transaction/coordinator/impl/MLTransactionMetadataStoreProvider.java b/pulsar-transaction/coordinator/src/main/java/org/apache/pulsar/transaction/coordinator/impl/MLTransactionMetadataStoreProvider.java
index bdf0d56..36b1958 100644
--- a/pulsar-transaction/coordinator/src/main/java/org/apache/pulsar/transaction/coordinator/impl/MLTransactionMetadataStoreProvider.java
+++ b/pulsar-transaction/coordinator/src/main/java/org/apache/pulsar/transaction/coordinator/impl/MLTransactionMetadataStoreProvider.java
@@ -23,6 +23,7 @@ import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
import org.apache.bookkeeper.mledger.ManagedLedgerFactory;
import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.transaction.coordinator.TransactionCoordinatorID;
+import org.apache.pulsar.transaction.coordinator.TransactionLog;
import org.apache.pulsar.transaction.coordinator.TransactionMetadataStore;
import org.apache.pulsar.transaction.coordinator.TransactionMetadataStoreProvider;
import org.apache.pulsar.transaction.coordinator.TransactionRecoverTracker;
@@ -44,16 +45,10 @@ public class MLTransactionMetadataStoreProvider implements TransactionMetadataSt
ManagedLedgerConfig managedLedgerConfig,
TransactionTimeoutTracker timeoutTracker,
TransactionRecoverTracker recoverTracker) {
- TransactionMetadataStore transactionMetadataStore;
- try {
- transactionMetadataStore =
- new MLTransactionMetadataStore(transactionCoordinatorId,
- new MLTransactionLogImpl(transactionCoordinatorId,
- managedLedgerFactory, managedLedgerConfig), timeoutTracker, recoverTracker);
- } catch (Exception e) {
- log.error("MLTransactionMetadataStore init fail", e);
- return FutureUtil.failedFuture(e);
- }
- return CompletableFuture.completedFuture(transactionMetadataStore);
+ MLTransactionLogImpl txnLog = new MLTransactionLogImpl(transactionCoordinatorId,
+ managedLedgerFactory, managedLedgerConfig);
+
+ return txnLog.initialize().thenApply(__ ->
+ new MLTransactionMetadataStore(transactionCoordinatorId, txnLog, timeoutTracker, recoverTracker));
}
}
\ No newline at end of file
diff --git a/pulsar-transaction/coordinator/src/test/java/org/apache/pulsar/transaction/coordinator/MLTransactionMetadataStoreTest.java b/pulsar-transaction/coordinator/src/test/java/org/apache/pulsar/transaction/coordinator/MLTransactionMetadataStoreTest.java
index 5316470..e9b3e0c 100644
--- a/pulsar-transaction/coordinator/src/test/java/org/apache/pulsar/transaction/coordinator/MLTransactionMetadataStoreTest.java
+++ b/pulsar-transaction/coordinator/src/test/java/org/apache/pulsar/transaction/coordinator/MLTransactionMetadataStoreTest.java
@@ -67,6 +67,7 @@ public class MLTransactionMetadataStoreTest extends MockedBookKeeperTestCase {
TransactionCoordinatorID transactionCoordinatorID = new TransactionCoordinatorID(1);
MLTransactionLogImpl mlTransactionLog = new MLTransactionLogImpl(transactionCoordinatorID, factory,
new ManagedLedgerConfig());
+ mlTransactionLog.initialize().join();
MLTransactionMetadataStore transactionMetadataStore =
new MLTransactionMetadataStore(transactionCoordinatorID, mlTransactionLog,
new TransactionTimeoutTrackerImpl(), new TransactionRecoverTrackerImpl());
@@ -138,6 +139,7 @@ public class MLTransactionMetadataStoreTest extends MockedBookKeeperTestCase {
managedLedgerConfig.setMaxEntriesPerLedger(3);
MLTransactionLogImpl mlTransactionLog = new MLTransactionLogImpl(transactionCoordinatorID, factory,
managedLedgerConfig);
+ mlTransactionLog.initialize().join();
MLTransactionMetadataStore transactionMetadataStore =
new MLTransactionMetadataStore(transactionCoordinatorID, mlTransactionLog,
new TransactionTimeoutTrackerImpl(), new TransactionRecoverTrackerImpl());
@@ -181,6 +183,7 @@ public class MLTransactionMetadataStoreTest extends MockedBookKeeperTestCase {
managedLedgerConfig.setMaxEntriesPerLedger(2);
MLTransactionLogImpl mlTransactionLog = new MLTransactionLogImpl(transactionCoordinatorID, factory,
managedLedgerConfig);
+ mlTransactionLog.initialize().join();
MLTransactionMetadataStore transactionMetadataStore =
new MLTransactionMetadataStore(transactionCoordinatorID, mlTransactionLog,
new TransactionTimeoutTrackerImpl(), new TransactionRecoverTrackerImpl());
@@ -220,10 +223,12 @@ public class MLTransactionMetadataStoreTest extends MockedBookKeeperTestCase {
transactionMetadataStore.closeAsync();
+ MLTransactionLogImpl txnLog2 = new MLTransactionLogImpl(transactionCoordinatorID, factory,
+ new ManagedLedgerConfig());
+ txnLog2.initialize().join();
MLTransactionMetadataStore transactionMetadataStoreTest =
new MLTransactionMetadataStore(transactionCoordinatorID,
- new MLTransactionLogImpl(transactionCoordinatorID, factory,
- new ManagedLedgerConfig()), new TransactionTimeoutTrackerImpl(), new TransactionRecoverTrackerImpl());
+ txnLog2, new TransactionTimeoutTrackerImpl(), new TransactionRecoverTrackerImpl());
while (true) {
if (checkReplayRetryCount > 6) {
@@ -285,6 +290,7 @@ public class MLTransactionMetadataStoreTest extends MockedBookKeeperTestCase {
TransactionCoordinatorID transactionCoordinatorID = new TransactionCoordinatorID(1);
MLTransactionLogImpl mlTransactionLog = new MLTransactionLogImpl(transactionCoordinatorID, factory,
new ManagedLedgerConfig());
+ mlTransactionLog.initialize().join();
MLTransactionMetadataStore transactionMetadataStore =
new MLTransactionMetadataStore(transactionCoordinatorID, mlTransactionLog,
new TransactionTimeoutTrackerImpl(), new TransactionRecoverTrackerImpl());
@@ -347,6 +353,7 @@ public class MLTransactionMetadataStoreTest extends MockedBookKeeperTestCase {
TransactionCoordinatorID transactionCoordinatorID = new TransactionCoordinatorID(1);
MLTransactionLogImpl mlTransactionLog = new MLTransactionLogImpl(transactionCoordinatorID, factory,
new ManagedLedgerConfig());
+ mlTransactionLog.initialize().join();
MLTransactionMetadataStore transactionMetadataStore =
new MLTransactionMetadataStore(transactionCoordinatorID, mlTransactionLog,
new TransactionTimeoutTrackerImpl(), new TransactionRecoverTrackerImpl());
@@ -364,6 +371,7 @@ public class MLTransactionMetadataStoreTest extends MockedBookKeeperTestCase {
mlTransactionLog = new MLTransactionLogImpl(transactionCoordinatorID, factory,
new ManagedLedgerConfig());
+ mlTransactionLog.initialize().join();
transactionMetadataStore =
new MLTransactionMetadataStore(transactionCoordinatorID, mlTransactionLog,
new TransactionTimeoutTrackerImpl(), new TransactionRecoverTrackerImpl());
@@ -381,6 +389,7 @@ public class MLTransactionMetadataStoreTest extends MockedBookKeeperTestCase {
TransactionCoordinatorID transactionCoordinatorID = new TransactionCoordinatorID(1);
MLTransactionLogImpl mlTransactionLog = new MLTransactionLogImpl(transactionCoordinatorID, factory,
new ManagedLedgerConfig());
+ mlTransactionLog.initialize().join();
MLTransactionMetadataStore transactionMetadataStore =
new MLTransactionMetadataStore(transactionCoordinatorID, mlTransactionLog,
new TransactionTimeoutTrackerImpl(), new TransactionRecoverTrackerImpl());