Skip to content

branch-4.0: [fix](test) stabilize BDB unmatched transaction test #65758

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
shuke987 wants to merge 1 commit into
base: branch-4.0
Choose a base branch
from
+43 −25
Open

branch-4.0: [fix](test) stabilize BDB unmatched transaction test #65758

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
68 changes: 43 additions & 25 deletions fe/fe-core/src/test/java/org/apache/doris/journal/bdbje/BDBEnvironmentTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,16 +36,17 @@
import com.sleepycat.je.rep.ReplicatedEnvironment;
import com.sleepycat.je.rep.impl.RepImpl;
import com.sleepycat.je.rep.util.ReplicationGroupAdmin;
import mockit.Expectations;
import mockit.Mock;
import mockit.MockUp;
import org.apache.commons.io.FileUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.awaitility.Awaitility;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.RepeatedTest;
import org.mockito.Mockito;

import java.io.File;
import java.io.IOException;
Expand All @@ -60,6 +61,7 @@
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

public class BDBEnvironmentTest {
private static final Logger LOG = LogManager.getLogger(BDBEnvironmentTest.class);
Expand Down Expand Up @@ -125,6 +127,19 @@ private byte[] randomBytes() {
return byteArray;
}

private void assertDatabaseValueEventually(BDBEnvironment environment, String dbName,
DatabaseEntry key, DatabaseEntry expectedValue) {
Awaitility.await().atMost(30, TimeUnit.SECONDS).untilAsserted(() -> {
Assertions.assertEquals(1, environment.getDatabaseNames().size());
Database database = environment.openDatabase(dbName);
Assertions.assertNotNull(database);
DatabaseEntry actualValue = new DatabaseEntry();
Assertions.assertEquals(OperationStatus.SUCCESS,
database.get(null, key, actualValue, LockMode.DEFAULT));
Assertions.assertArrayEquals(expectedValue.getData(), actualValue.getData());
});
}

private static DatabaseEntry longToEntry(long value) {
DatabaseEntry key = new DatabaseEntry();
TupleBinding<Long> idBinding = TupleBinding.getPrimitiveBinding(Long.class);
Expand Down Expand Up @@ -689,34 +704,36 @@ public void testReadTxnIsNotMatched() throws Exception {
continue;
}

Assertions.assertEquals(1, entryPair.first.getDatabaseNames().size());
Database followerDb = entryPair.first.openDatabase(beginDbName);
DatabaseEntry readValue = new DatabaseEntry();
Assertions.assertEquals(OperationStatus.SUCCESS, followerDb.get(null, key, readValue, LockMode.DEFAULT));
Assertions.assertEquals(new String(value.getData()), new String(readValue.getData()));
assertDatabaseValueEventually(entryPair.first, beginDbName, key, value);
}

long count = masterDb.count();
DatabaseEntry localCommitKey = new DatabaseEntry(new byte[]{1, 2, 3});
DatabaseEntry localCommitValue = new DatabaseEntry(new byte[]{4, 5, 6});

ReplicatedEnvironment replicatedEnvironment = masterPair.first.getReplicatedEnvironment();
Field envImplField = ReplicatedEnvironment.class.getDeclaredField("repEnvironmentImpl");
envImplField.setAccessible(true);
RepImpl impl = (RepImpl) envImplField.get(masterPair.first.getReplicatedEnvironment());
RepImpl impl = (RepImpl) envImplField.get(replicatedEnvironment);
Assertions.assertNotNull(impl);

new Expectations(impl) {{
// Below method will replicate log item to followers.
impl.registerVLSN(withNotNull());
// Below method will wait until the logs are replicated.
impl.postLogCommitHook(withNotNull(), withNotNull());
result = new InsufficientAcksException("mocked");
}};

long count = masterDb.count();
final Database oldMasterDb = masterDb;
Assertions.assertThrows(InsufficientAcksException.class, () -> {
// Since this key is not replicated to any replicas, it should not be read.
DatabaseEntry k = new DatabaseEntry(new byte[]{1, 2, 3});
DatabaseEntry v = new DatabaseEntry(new byte[]{4, 5, 6});
oldMasterDb.put(null, k, v);
});
Field environmentImplField = com.sleepycat.je.Environment.class.getDeclaredField("environmentImpl");
environmentImplField.setAccessible(true);
RepImpl implSpy = Mockito.spy(impl);
Mockito.doThrow(new InsufficientAcksException("mocked"))
.when(implSpy).postLogCommitHook(Mockito.any(), Mockito.any());

try {
envImplField.set(replicatedEnvironment, implSpy);
environmentImplField.set(replicatedEnvironment, implSpy);
final Database oldMasterDb = masterDb;
// Simulate an acknowledgement failure after the transaction is committed locally.
Assertions.assertThrows(InsufficientAcksException.class,
() -> oldMasterDb.put(null, localCommitKey, localCommitValue));
} finally {
envImplField.set(replicatedEnvironment, impl);
environmentImplField.set(replicatedEnvironment, impl);
}

LOG.info("close old master {} | {}", masterPair.second.name, masterPair.second.dir);
masterDb.close();
Expand All @@ -739,9 +756,10 @@ public void testReadTxnIsNotMatched() throws Exception {
// The local commit txn is readable!!!
Assertions.assertEquals(count + 1, masterDb.count());

key = new DatabaseEntry(new byte[]{1, 2, 3});
DatabaseEntry readValue = new DatabaseEntry();
Assertions.assertEquals(OperationStatus.SUCCESS, masterDb.get(null, key, readValue, LockMode.DEFAULT));
Assertions.assertEquals(OperationStatus.SUCCESS,
masterDb.get(null, localCommitKey, readValue, LockMode.DEFAULT));
Assertions.assertArrayEquals(localCommitValue.getData(), readValue.getData());
} finally {
followersInfo.stream().forEach(entryPair -> {
entryPair.first.close();
Expand Down
Loading