[fix][test] Fix the bug caused by unload topic and compaction task failure after task is triggered #21634

Denovo1998 · 2023-11-28T16:16:33Z

Motivation

Because of #21536 . The consumer with the subscription name __compaction can not reconnect. Therefore, the compaction task triggered before the topic unload fails. In this case, a compaction task can be triggered again after unload.

Modifications

After unload the topic, make sure that the topic has been initialized correctly. Then trigger compaction again after previous task failed.

Verifying this change

Make sure that the change passes the CI checks.

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

(or)

This change is already covered by existing tests, such as (please describe tests).

(or)

This change added tests and can be verified as follows:

(example:)

Added integration tests for end-to-end deployment with large payloads (10MB)
Extended integration test for recovery after broker failure

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

Documentation

doc
doc-required
doc-not-needed
doc-complete

Matching PR in forked repository

PR in forked repository: Denovo1998#5

pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java

Denovo1998 · 2023-11-29T16:58:12Z

@coderzc @Technoboy- CI checks reported an error just now and has been fixed now. Please check again.

pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java

coderzc · 2023-11-30T05:35:27Z

@Denovo1998 After looking at the code, I found that getCompactedTopicContext method will wait for compactedTopicContext to be created, so we don’t seem to need to wait for compactedLedger to be created in the test. I think we just need to wait for topic initialization to complete.

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java

Lines 298 to 304 in 4e7bff7

    
               /** 
        
                * Getter for CompactedTopicContext. 
        
                * @return CompactedTopicContext 
        
                */ 
        
               public Optional<CompactedTopicContext> getCompactedTopicContext() throws ExecutionException, InterruptedException { 
        
                   return compactedTopicContext == null ? Optional.empty() : Optional.of(compactedTopicContext.get()); 
        
               }

Denovo1998 · 2023-11-30T06:02:11Z

@Denovo1998 After looking at the code, I found that getCompactedTopicContext method will wait for compactedTopicContext to be created, so we don’t seem to need to wait for compactedLedger to be created in the test. I think we just need to wait for topic initialization to complete.

After the topic initialization is complete, we also wait for the CompactedTopicContext to be created. Because the creation of CompactedTopicContext requires the creation of compactedLedger, otherwise ledgerId will be -1, and an error will be reported if a ledger is not created after 10s. here

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/compaction/CompactedTopicImpl.java

Lines 239 to 253 in e1d06b5

    
           private static CompletableFuture<CompactedTopicContext> openCompactedLedger(BookKeeper bk, long id) { 
        
               CompletableFuture<LedgerHandle> promise = new CompletableFuture<>(); 
        
               bk.asyncOpenLedger(id, 
        
                                  Compactor.COMPACTED_TOPIC_LEDGER_DIGEST_TYPE, 
        
                                  Compactor.COMPACTED_TOPIC_LEDGER_PASSWORD, 
        
                                  (rc, ledger, ctx) -> { 
        
                                      if (rc != BKException.Code.OK) { 
        
                                          promise.completeExceptionally(BKException.create(rc)); 
        
                                      } else { 
        
                                          promise.complete(ledger); 
        
                                      } 
        
                                  }, null); 
        
               return promise.thenApply((ledger) -> new CompactedTopicContext( 
        
                                                ledger, createCache(ledger, DEFAULT_STARTPOINT_CACHE_SIZE))); 
        
           }

coderzc · 2023-11-30T06:53:19Z

2023-11-30T05:30:45,771 - WARN  - [broker-client-shared-internal-executor-4522-1:PersistentTopic@3397] - [persistent://my-property/use/my-ns/testCompactionDuplicate] Compaction failure.
java.util.concurrent.CompletionException: org.apache.pulsar.client.api.PulsarClientException$ConnectException: Consumer connect fail! consumer state:Connecting
	at java.util.concurrent.CompletableFuture.encodeRelay(CompletableFuture.java:368) ~[?:?]
	at java.util.concurrent.CompletableFuture.completeRelay(CompletableFuture.java:377) ~[?:?]
	at java.util.concurrent.CompletableFuture$UniCompose.tryFire(CompletableFuture.java:1152) ~[?:?]
	at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:510) ~[?:?]
	at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2147) ~[?:?]
	at org.apache.pulsar.compaction.TwoPhaseCompactor.lambda$closeLedger$20(TwoPhaseCompactor.java:388) ~[classes/:?]
	at org.apache.bookkeeper.client.PulsarMockLedgerHandle.lambda$asyncClose$1(PulsarMockLedgerHandle.java:96) ~[testmocks-3.2.0-SNAPSHOT.jar:4.16.3]
	at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:863) ~[?:?]
	at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:841) ~[?:?]
	at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:482) ~[?:?]
	at org.apache.bookkeeper.common.util.SingleThreadExecutor.safeRunTask(SingleThreadExecutor.java:137) ~[bookkeeper-common-4.16.3.jar:4.16.3]
	at org.apache.bookkeeper.common.util.SingleThreadExecutor.run(SingleThreadExecutor.java:107) ~[bookkeeper-common-4.16.3.jar:4.16.3]
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) ~[netty-common-4.1.100.Final.jar:4.1.100.Final]
	at java.lang.Thread.run(Thread.java:840) ~[?:?]
Caused by: org.apache.pulsar.client.api.PulsarClientException$ConnectException: Consumer connect fail! consumer state:Connecting
	at org.apache.pulsar.client.impl.PersistentAcknowledgmentsGroupingTracker.doImmediateAck(PersistentAcknowledgmentsGroupingTracker.java:363) ~[pulsar-client-original-3.2.0-SNAPSHOT.jar:3.2.0-SNAPSHOT]
	at org.apache.pulsar.client.impl.PersistentAcknowledgmentsGroupingTracker.doCumulativeAck(PersistentAcknowledgmentsGroupingTracker.java:310) ~[pulsar-client-original-3.2.0-SNAPSHOT.jar:3.2.0-SNAPSHOT]
	at org.apache.pulsar.client.impl.PersistentAcknowledgmentsGroupingTracker.addAcknowledgment(PersistentAcknowledgmentsGroupingTracker.java:243) ~[pulsar-client-original-3.2.0-SNAPSHOT.jar:3.2.0-SNAPSHOT]
	at org.apache.pulsar.client.impl.PersistentAcknowledgmentsGroupingTracker.addAcknowledgment(PersistentAcknowledgmentsGroupingTracker.java:198) ~[pulsar-client-original-3.2.0-SNAPSHOT.jar:3.2.0-SNAPSHOT]
	at org.apache.pulsar.client.impl.ConsumerImpl.doAcknowledge(ConsumerImpl.java:566) ~[pulsar-client-original-3.2.0-SNAPSHOT.jar:3.2.0-SNAPSHOT]
	at org.apache.pulsar.client.impl.ConsumerBase.doAcknowledgeWithTxn(ConsumerBase.java:688) ~[pulsar-client-original-3.2.0-SNAPSHOT.jar:3.2.0-SNAPSHOT]
	at org.apache.pulsar.client.impl.RawReaderImpl.acknowledgeCumulativeAsync(RawReaderImpl.java:90) ~[classes/:3.2.0-SNAPSHOT]
	at org.apache.pulsar.compaction.TwoPhaseCompactor.lambda$phaseTwoSeekThenLoop$10(TwoPhaseCompactor.java:225) ~[classes/:?]
	at java.util.concurrent.CompletableFuture$UniCompose.tryFire(CompletableFuture.java:1150) ~[?:?]

@Denovo1998 I found the above error logs from the CI report. I think this is the root cause. I think we can trigger compaction again after topic unload. Maybe we can get the future of compaction from persistentTopic and judge whether compaction task has failed, then retrigger it.

Denovo1998 · 2023-11-30T08:24:02Z

2023-11-30T05:30:45,771 - WARN  - [broker-client-shared-internal-executor-4522-1:PersistentTopic@3397] - [persistent://my-property/use/my-ns/testCompactionDuplicate] Compaction failure.
java.util.concurrent.CompletionException: org.apache.pulsar.client.api.PulsarClientException$ConnectException: Consumer connect fail! consumer state:Connecting
	at java.util.concurrent.CompletableFuture.encodeRelay(CompletableFuture.java:368) ~[?:?]
	at java.util.concurrent.CompletableFuture.completeRelay(CompletableFuture.java:377) ~[?:?]
	at java.util.concurrent.CompletableFuture$UniCompose.tryFire(CompletableFuture.java:1152) ~[?:?]
	at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:510) ~[?:?]
	at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2147) ~[?:?]
	at org.apache.pulsar.compaction.TwoPhaseCompactor.lambda$closeLedger$20(TwoPhaseCompactor.java:388) ~[classes/:?]
	at org.apache.bookkeeper.client.PulsarMockLedgerHandle.lambda$asyncClose$1(PulsarMockLedgerHandle.java:96) ~[testmocks-3.2.0-SNAPSHOT.jar:4.16.3]
	at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:863) ~[?:?]
	at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:841) ~[?:?]
	at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:482) ~[?:?]
	at org.apache.bookkeeper.common.util.SingleThreadExecutor.safeRunTask(SingleThreadExecutor.java:137) ~[bookkeeper-common-4.16.3.jar:4.16.3]
	at org.apache.bookkeeper.common.util.SingleThreadExecutor.run(SingleThreadExecutor.java:107) ~[bookkeeper-common-4.16.3.jar:4.16.3]
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) ~[netty-common-4.1.100.Final.jar:4.1.100.Final]
	at java.lang.Thread.run(Thread.java:840) ~[?:?]
Caused by: org.apache.pulsar.client.api.PulsarClientException$ConnectException: Consumer connect fail! consumer state:Connecting
	at org.apache.pulsar.client.impl.PersistentAcknowledgmentsGroupingTracker.doImmediateAck(PersistentAcknowledgmentsGroupingTracker.java:363) ~[pulsar-client-original-3.2.0-SNAPSHOT.jar:3.2.0-SNAPSHOT]
	at org.apache.pulsar.client.impl.PersistentAcknowledgmentsGroupingTracker.doCumulativeAck(PersistentAcknowledgmentsGroupingTracker.java:310) ~[pulsar-client-original-3.2.0-SNAPSHOT.jar:3.2.0-SNAPSHOT]
	at org.apache.pulsar.client.impl.PersistentAcknowledgmentsGroupingTracker.addAcknowledgment(PersistentAcknowledgmentsGroupingTracker.java:243) ~[pulsar-client-original-3.2.0-SNAPSHOT.jar:3.2.0-SNAPSHOT]
	at org.apache.pulsar.client.impl.PersistentAcknowledgmentsGroupingTracker.addAcknowledgment(PersistentAcknowledgmentsGroupingTracker.java:198) ~[pulsar-client-original-3.2.0-SNAPSHOT.jar:3.2.0-SNAPSHOT]
	at org.apache.pulsar.client.impl.ConsumerImpl.doAcknowledge(ConsumerImpl.java:566) ~[pulsar-client-original-3.2.0-SNAPSHOT.jar:3.2.0-SNAPSHOT]
	at org.apache.pulsar.client.impl.ConsumerBase.doAcknowledgeWithTxn(ConsumerBase.java:688) ~[pulsar-client-original-3.2.0-SNAPSHOT.jar:3.2.0-SNAPSHOT]
	at org.apache.pulsar.client.impl.RawReaderImpl.acknowledgeCumulativeAsync(RawReaderImpl.java:90) ~[classes/:3.2.0-SNAPSHOT]
	at org.apache.pulsar.compaction.TwoPhaseCompactor.lambda$phaseTwoSeekThenLoop$10(TwoPhaseCompactor.java:225) ~[classes/:?]
	at java.util.concurrent.CompletableFuture$UniCompose.tryFire(CompletableFuture.java:1150) ~[?:?]

@Denovo1998 I found the above error logs from the CI report. I think this is the root cause. I think we can trigger compaction again after topic unload. Maybe we can get the future of compaction from persistentTopic and judge whether compaction task has failed, then retrigger it.

@coderzc I get it! So the cause is that triggerCompactor didn't succeed, not that compactionLedger creation timed out.
We need to unload the topic, make sure the topic properly initialized, and invoke persistentTopic.compactionStatus(), if success is to continue. If that fails, call triggerCompactor again.
If my description is correct, I will modify the solution description and code of PR after work.

coderzc · 2023-11-30T08:37:03Z

@coderzc I get it! So the cause is that triggerCompactor didn't succeed, not that compactionLedger creation timed out.
We need to unload the topic, make sure the topic properly initialized, and invoke persistentTopic.compactionStatus(), if success is to continue. If that fails, call triggerCompactor again.
If my description is correct, I will modify the solution description and code of PR after work.

@Denovo1998 Yes, I think so.

Denovo1998 · 2023-11-30T13:56:07Z

@coderzc Something strange.

1. Call persistentTopic.compactionStatus() after unload topic, longRunningProcessStatus always NOT_RUN

2. Even if the topic already exists, the compactedTopicContext is not created correctly.

The problem still seems to be the compactedTopicContext, The compactedLedger hasn't been created yet
(org.apache.pulsar.compaction.CompactedTopicImpl#openCompactedLedger).

PersistentTopicInternalStats internalStats = admin.topics().getInternalStats(topic, false);

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

Lines 2455 to 2469 in 3377003

    
           // Add ledger info for compacted topic ledger if exist. 
        
           LedgerInfo info = new LedgerInfo(); 
        
           info.ledgerId = -1; 
        
           info.entries = -1; 
        
           info.size = -1; 
        
           Optional<CompactedTopicContext> compactedTopicContext = getCompactedTopicContext(); 
        
           if (compactedTopicContext.isPresent()) { 
        
               CompactedTopicContext ledgerContext = compactedTopicContext.get(); 
        
               info.ledgerId = ledgerContext.getLedger().getId(); 
        
               info.entries = ledgerContext.getLedger().getLastAddConfirmed() + 1; 
        
               info.size = ledgerContext.getLedger().getLength(); 
        
           } 
        
           stats.compactedLedger = info;

3. The error you sent before should be this issue. Does the consumer with the subscription name __compaction not reconnect?

#21536

coderzc · 2023-11-30T14:45:21Z

@Denovo1998

Call persistentTopic.compactionStatus() after unload topic, longRunningProcessStatus always NOT_RUN

Because after unloading, persistentTopic.compactionStatus() will be reset, so we cannot get the running status of the previous task through it. We can get and retain the CompletableFuture object of previous task (persistentTopic .currentCompaction).

Denovo1998 · 2023-11-30T14:50:18Z

@coderzc Yes, I just read the code and got it. Let's look at the second and third questions.

coderzc · 2023-11-30T14:53:43Z

Even if the topic already exists, the compactedTopicContext is not created correctly.

@Denovo1998 Maybe the compaction task isn't finished yet? CompactedTopicContext is created only after the compaction task is completed.

The error you sent before should be this issue. Does the consumer with the subscription name __compaction not reconnect?

@Denovo1998 Yes, the reader of compaction may be reconnecting when acknowledging message.

Denovo1998 · 2023-11-30T15:00:25Z

@coderzc Therefore, the current code in this pr waits for the CompactedTopicContext to be created successfully, which has solved this issue.
As for the error report in number 3, it seems to be the problem of issue21536?

coderzc · 2023-11-30T15:13:11Z

As for the error report in number 3, it seems to be the problem of #21536?

why? we also need to fix it in StrategicCompactionTest.testCompactionDuplicate. just retrigger compaction if the previous task failed after topic was unloaded.

Denovo1998 · 2023-11-30T18:31:29Z

we also need to fix it in StrategicCompactionTest.testCompactionDuplicate. just retrigger compaction if the previous task failed after topic was unloaded.

@coderzc Do you mean to create a new test case in org.apache.pulsar.compaction.StrategicCompactionTest?

All has been changed. Please check again.

coderzc · 2023-12-01T02:29:13Z

we also need to fix it in StrategicCompactionTest.testCompactionDuplicate. just retrigger compaction if the previous task failed after topic was unloaded.

@coderzc Do you mean to create a new test case in org.apache.pulsar.compaction.StrategicCompactionTest?

All has been changed. Please check again.

Oh, Don't need to do that., StrategicCompactionTest already extends from CompactionTest.

Denovo1998 · 2023-12-01T03:42:39Z

Oh, Don't need to do that., StrategicCompactionTest already extends from CompactionTest.

@coderzc I'm sorry I misunderstood what you said. I am still learning the code structure of pulsar, and I am not particularly familiar with it. The modification to the org.apache.pulsar.com paction.StrategicCompactionTest has been removed. Please check again.

pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java

coderzc

LGTM

coderzc · 2023-12-01T09:03:24Z

/pulsarbot rerun-failure-checks

codecov-commenter · 2023-12-01T10:02:49Z

Codecov Report

Merging #21634 (5ae340f) into master (e1d06b5) will increase coverage by 36.49%.
Report is 12 commits behind head on master.
The diff coverage is n/a.

Additional details and impacted files

@@              Coverage Diff              @@
##             master   #21634       +/-   ##
=============================================
+ Coverage     36.83%   73.33%   +36.49%     
- Complexity      376    32740    +32364     
=============================================
  Files          1715     1893      +178     
  Lines        131097   140730     +9633     
  Branches      14318    15500     +1182     
=============================================
+ Hits          48288   103201    +54913     
+ Misses        76424    29421    -47003     
- Partials       6385     8108     +1723

Flag	Coverage Δ
inttests	`24.15% <ø> (-0.09%)`	⬇️
systests	`24.66% <ø> (-0.26%)`	⬇️
unittests	`72.66% <ø> (+40.79%)`	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

see 1460 files with indirect coverage changes

wait for newCompactedLedger

4812580

github-actions bot added the doc-not-needed Your PR changes do not impact docs label Nov 28, 2023

Denovo1998 mentioned this pull request Nov 28, 2023

[fix][broker] Fix the bug that the CompactedLedger is used before the newCompactedLedger done Denovo1998/pulsar#5

Closed

coderzc reviewed Nov 29, 2023

View reviewed changes

pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java Outdated Show resolved Hide resolved

coderzc changed the title ~~[fix][broker] Fix the bug that the CompactedLedger is used before the newCompactedLedger done~~ [fix][test] Fix the bug that the CompactedLedger is used before the newCompactedLedger done Nov 29, 2023

Technoboy- added this to the 3.2.0 milestone Nov 29, 2023

Technoboy- assigned Denovo1998 Nov 29, 2023

Technoboy- added the ready-to-test label Nov 29, 2023

wait for newCompactedLedger after topic unload

f5e82cd

Technoboy- added area/test type/flaky-tests labels Nov 29, 2023

Technoboy- approved these changes Nov 29, 2023

View reviewed changes

coderzc approved these changes Nov 29, 2023

View reviewed changes

Denovo1998 added 4 commits November 29, 2023 22:00

keep the original code

8c5dcd3

wait for persistentTopic present

9e673ef

wait for persistentTopic present after unload topic

c1fc1db

wait for topic exists

600ce0f

Technoboy- reviewed Nov 30, 2023

View reviewed changes

pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java Outdated Show resolved Hide resolved

Denovo1998 added 2 commits December 1, 2023 02:08

trigger compaction again after previous task failed

312b650

test compactionDuplicate with StrategicCompaction

f7415b5

remove testCompactionDuplicateWithStrategicCompaction

b413320

coderzc reviewed Dec 1, 2023

View reviewed changes

pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionTest.java Show resolved Hide resolved

remove processing that does not fail the previous task

310760e

Denovo1998 changed the title ~~[fix][test] Fix the bug that the CompactedLedger is used before the newCompactedLedger done~~ [fix][test] Fix the bug caused by unload topic and compaction task failure after task is triggered Dec 1, 2023

make sure previous compaction task success

5ae340f

coderzc approved these changes Dec 1, 2023

View reviewed changes

coderzc merged commit 2bf1354 into apache:master Dec 1, 2023

Denovo1998 deleted the wait_for_newCompactedLedger branch December 1, 2023 14:23

[fix][test] Fix the bug caused by unload topic and compaction task failure after task is triggered #21634

[fix][test] Fix the bug caused by unload topic and compaction task failure after task is triggered #21634

Uh oh!

Conversation

Denovo1998 commented Nov 28, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Motivation

Modifications

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Matching PR in forked repository

Uh oh!

Uh oh!

Denovo1998 commented Nov 29, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

coderzc commented Nov 30, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Denovo1998 commented Nov 30, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

coderzc commented Nov 30, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Denovo1998 commented Nov 30, 2023

Uh oh!

coderzc commented Nov 30, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Denovo1998 commented Nov 30, 2023

Uh oh!

coderzc commented Nov 30, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Denovo1998 commented Nov 30, 2023

Uh oh!

coderzc commented Nov 30, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Denovo1998 commented Nov 30, 2023

Uh oh!

coderzc commented Nov 30, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Denovo1998 commented Nov 30, 2023

Uh oh!

coderzc commented Dec 1, 2023

Uh oh!

Denovo1998 commented Dec 1, 2023

Uh oh!

Uh oh!

coderzc left a comment

Choose a reason for hiding this comment

Uh oh!

coderzc commented Dec 1, 2023

Uh oh!

codecov-commenter commented Dec 1, 2023

Codecov Report

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

Denovo1998 commented Nov 28, 2023 •

edited

Loading

Denovo1998 commented Nov 29, 2023 •

edited

Loading

coderzc commented Nov 30, 2023 •

edited

Loading

Denovo1998 commented Nov 30, 2023 •

edited

Loading

coderzc commented Nov 30, 2023 •

edited

Loading

coderzc commented Nov 30, 2023 •

edited

Loading

coderzc commented Nov 30, 2023 •

edited

Loading

coderzc commented Nov 30, 2023 •

edited

Loading

coderzc commented Nov 30, 2023 •

edited

Loading