dimas-b commented on code in PR #4552:
URL: https://github.com/apache/polaris/pull/4552#discussion_r3312653498
##########
persistence/nosql/persistence/impl/src/testFixtures/java/org/apache/polaris/persistence/nosql/impl/commits/BaseTestCommitterImpl.java:
##########
@@ -50,6 +54,19 @@ public abstract class BaseTestCommitterImpl {
@PolarisPersistence(fastRetries = true)
protected Persistence persistence;
+ @Test
+ public void exclusiveSynchronizerDoesNotReleaseUnacquiredPermit() {
+ var sync =
+ ExclusiveCommitSynchronizer.forKey(
+ persistence.realmId(), "exclusiveSynchronizer-" +
persistence.generateId());
+
+ soft.assertThat(sync.before(0L)).isTrue();
+ soft.assertThat(sync.before(1L)).isFalse();
+ sync.after();
Review Comment:
TBH, I do not see how this test makes sure that
`exclusiveSynchronizerDoesNotReleaseUnacquiredPermit` 🤔
##########
persistence/nosql/persistence/impl/src/testFixtures/java/org/apache/polaris/persistence/nosql/impl/commits/BaseTestCommitterImpl.java:
##########
@@ -269,6 +286,75 @@ public void simpleImmediatelySuccessfulCommit(TestInfo
testInfo) throws Exceptio
.containsExactly(anotherObj1.withNumParts(1),
anotherObj2.withNumParts(1));
}
+ @Test
+ public void synchronizingLocallySerializesConcurrentCommits(TestInfo
testInfo) throws Exception {
+ var initialObj =
+ persistence.write(
+ CommitTestObj.builder()
+ .id(persistence.generateId())
+ .text("initial")
+ .seq(1)
+ .tail(new long[0])
+ .build(),
+ CommitTestObj.class);
+ var referenceName = testInfo.getTestMethod().orElseThrow().getName();
+
+ persistence.write(initialObj, CommitTestObj.class);
+ persistence.createReference(referenceName,
Optional.of(objRef(initialObj)));
+
+ var committer =
+ persistence
+ .createCommitter(referenceName, CommitTestObj.class, String.class)
+ .synchronizingLocally();
+ var firstEntered = new CountDownLatch(1);
+ var releaseFirst = new CountDownLatch(1);
+ var secondEntered = new CountDownLatch(1);
+ var firstReleased = new AtomicBoolean();
+ var executor = Executors.newFixedThreadPool(2);
+ try {
+ var first =
+ executor.submit(
+ () ->
+ committer.commit(
+ (state, refObjSupplier) -> {
+ firstEntered.countDown();
+ try {
+ if (!releaseFirst.await(5, TimeUnit.SECONDS)) {
+ throw new AssertionError("Timed out waiting to
release first commit");
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ firstReleased.set(true);
+ return state.commitResult(
+ "first", CommitTestObj.builder().text("first"),
refObjSupplier.get());
+ }));
+ soft.assertThat(firstEntered.await(5, TimeUnit.SECONDS)).isTrue();
+
+ var second =
+ executor.submit(
+ () ->
+ committer.commit(
+ (state, refObjSupplier) -> {
+ secondEntered.countDown();
+ if (!firstReleased.get()) {
+ throw new AssertionError(
+ "Second synchronized commit started before first
commit finished");
Review Comment:
Not sure this assert is conclusive 🤔 It might as well (and likely) hold true
by coincidence 🤔
I think the test would be more robust if `firstReleased.set(true)` happened
after `commitResult()` and an extra (not too long) delay.
Ideally we should submit a third task after "second" to the executor
directly (free from the `Semaphore`) and validate that it runs in-between first
(before first is released) and second.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
