[improve][broker] PIP-192 Added ServiceUnitStateCompactionStrategy #19045
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
heesung-sn
changed the title
heesung-sn
force-pushed
the
pip-192-compact-service-unit-state
branch
from
d4b4e85 to
3e7c2bf
Compare
congbobo184
requested review from
codelipenghui,
gaoran10,
Demogorgon314,
merlimat,
eolivelli,
lhotari and
nicoloboschi
.../apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateCompactionStrategy.java
Show resolved
Hide resolved
pulsar-broker/src/test/java/org/apache/pulsar/compaction/ServiceUnitStateCompactionTest.java
Outdated
Show resolved
Hide resolved
…s in ServiceUnitStateCompactionTest
heesung-sn
force-pushed
the
pip-192-compact-service-unit-state
branch
from
271bee2 to
e348be3
Compare
Demogorgon314
approved these changes
Jan 19, 2023
Codecov Report
@@ Coverage Diff @@
## master #19045 +/- ##
=============================================
+ Coverage 47.20% 60.71% +13.51%
- Complexity 10645 26081 +15436
=============================================
Files 709 1884 +1175
Lines 69421 137392 +67971
Branches 7449 15106 +7657
=============================================
+ Hits 32769 83424 +50655
- Misses 32984 46262 +13278
- Partials 3668 7706 +4038
Flags with carried forward coverage won't be shown. Click here to find out more.
|
|
@Technoboy- Hi, can you take a look again? |
eolivelli
reviewed
Jan 23, 2023
| private boolean checkBrokers = true; | ||
|
|
||
| public ServiceUnitStateCompactionStrategy() { | ||
| schema = JSONSchema.of(ServiceUnitStateData.class); |
eolivelli
reviewed
Jan 23, 2023
| @@ -385,6 +385,7 @@ private <T> void phaseTwoLoop(String topic, Iterator<Message<T>> reader, | |||
| promise.completeExceptionally(e); | |||
| return; | |||
| } | |||
| outstanding.release(MAX_OUTSTANDING); | |||
There was a problem hiding this comment.
should we move this after promise.complete(null); ?
There was a problem hiding this comment.
No, I found a deadlock if we don't release this semaphore before completing the future.
deadLock example
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import org.junit.Test;
public class SemaphoreTest {
@Test
public void deadlockTest() {
Semaphore sm = new Semaphore(2);
ExecutorService executor = Executors.newSingleThreadExecutor();
CompletableFuture<String> future = new CompletableFuture();
CompletableFuture.supplyAsync(() -> {
System.out.println("starting");
try {
sm.acquire(2);
System.out.println("acquired 2");
} catch (InterruptedException e) {
}
try {
Thread.sleep(1000*5);
} catch (InterruptedException e) {
}
System.out.println("return!");
future.complete(""); // hangs here.
System.out.println("proceeding");
sm.release(2);
System.out.println("released 2");
return "";
}, executor);
future.thenCompose(x -> {
try {
System.out.println("acquiring one!");
sm.acquire(); // hangs here.
System.out.println("acquired 1");
sm.release();
} catch (InterruptedException e) {
}
return CompletableFuture.completedFuture("");
}).join();
}
}
eolivelli
reviewed
Jan 23, 2023
| compactionScheduler = Executors.newSingleThreadScheduledExecutor( | ||
| new ThreadFactoryBuilder().setNameFormat("compaction-%d").setDaemon(true).build()); | ||
| bk = pulsar.getBookKeeperClientFactory().create(this.conf, null, null, Optional.empty(), null); | ||
| schema = JSONSchema.of(ServiceUnitStateData.class); |
heesung-sn
force-pushed
the
pip-192-compact-service-unit-state
branch
from
e0c7cfb to
bf665d3
Compare
Technoboy-
reviewed
Jan 28, 2023
| log.error("Failed cleaning the ownership serviceUnit:{}, stateData:{}.", | ||
| serviceUnit, stateData); | ||
| serviceUnitTombstoneErrorCnt.incrementAndGet(); | ||
| } | ||
| }); | ||
| serviceUnitTombstoneCnt.incrementAndGet(); |
There was a problem hiding this comment.
Could you help explain why do we need modify here ?
we update serviceUnitTombstoneErrorCnt and serviceUnitTombstoneCnt in the async method. these values could both be 0. and let line 643 pass.
There was a problem hiding this comment.
It relies on producer.flush() to persist all outstanding messages before returning this call.
Yes, because these values could be 0, I updated this code path here. Also, I further cleaned the metrics computation code to be cleaner.
heesung-sn
force-pushed
the
pip-192-compact-service-unit-state
branch
from
3618416 to
cb61a15
Compare
heesung-sn
force-pushed
the
pip-192-compact-service-unit-state
branch
from
cb61a15 to
c818108
Compare
|
@Technoboy- ping |
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
Show resolved
Hide resolved
| @@ -1563,6 +1571,8 @@ public void checkCompaction() { | |||
| } | |||
|
|
|||
| if (backlogEstimate > compactionThreshold) { | |||
| log.info("topic:{} backlogEstimate:{} is bigger than compactionThreshold:{}. Triggering compaction", | |||
There was a problem hiding this comment.
It should be a debug-level log.
There was a problem hiding this comment.
I thought this topic compaction log is useful. Could you share your concerns about this?
There was a problem hiding this comment.
Suppose you have many topics, like 50k topics per broker. By default, we will have 50k logs per minute. And we already have logs after the compaction task started.
There was a problem hiding this comment.
Ok. I will make this debug lvl.
codelipenghui
approved these changes
Jan 31, 2023
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Master Issue: #16691
Motivation
This PR adds ServiceUnitStateCompactionStrategy.
Modifications
For the PIP-192, this PR adds the
ServiceUnitStateCompactionStrategyand its unit tests.Also, this PR updates the related classes to enable ServiceUnitStateCompactionStrategy:
strategicCompactormember in PulsarService.ServiceUnitStateCompactionStrategyfor thetableviewmember in ServiceUnitStateChannelImpl.strategicCompactionMapmember in PersistentTopicVerifying this change
This change added tests and can be verified as follows:
Does this pull request potentially affect one of the following parts:
If the box was checked, please highlight the changes
Documentation
docdoc-requireddoc-not-neededdoc-completeWe will have separate PRs to update the Doc later.
Matching PR in forked repository
PR in forked repository: heesung-sn#18