[fix][broker] Fix the markdelete position does not move forward when isAutoSkipNonRecoverableData=true and individual ack #17751

lordcheng10 · 2022-09-20T16:44:03Z

Motivation

When isAutoSkipNonRecoverableData=true and individual ack, the markdelete position does not move forward.

pulsar/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpReadEntry.java

Lines 100 to 114 in 72dd01b

 } else if (cursor.config.isAutoSkipNonRecoverableData() && exception instanceof NonRecoverableLedgerException) { 

 log.warn("[{}][{}] read failed from ledger at position:{} : {}", cursor.ledger.getName(), cursor.getName(), 

 readPosition, exception.getMessage()); 

 // try to find and move to next valid ledger 

 final Position nexReadPosition = cursor.getNextLedgerPosition(readPosition.getLedgerId()); 

 // fail callback if it couldn't find next valid ledger 

 if (nexReadPosition == null) { 

 callback.readEntriesFailed(exception, ctx); 

 cursor.ledger.mbean.recordReadEntriesError(); 

 recycle(); 

 return; 

 } 

 updateReadPosition(nexReadPosition); 

 checkReadCompletion(); 

 } else {

Modifications

Add a skipped variable in EntryImpl to record whether the entry is automatically skipped;
When pushing entrys to the consumer, automatically ack the skipped entrys;

Documentation

doc-required
(Your PR needs to update docs and you will update later)
doc-not-needed
(Please explain why)
doc
(Your PR contains doc changes)
doc-complete
(Docs have been already added)

Matching PR in forked repository

PR in forked repository: lordcheng10#19

lordcheng10 · 2022-09-20T17:20:57Z

@eolivelli @codelipenghui @Technoboy- PTAL,thanks!

eolivelli · 2022-09-25T12:44:31Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java

+ */
+ public List<Entry> filterAndAcknowledgeSkippedEntry(List<Entry> entries) {
+ List<Position> skippedPositions = new ArrayList<>();
+ List<Entry> filterEntries = Lists.newArrayList(Collections2.filter(entries, entry -> {


I wonder if we can move this into filterEntriesForConsumer? This way we save some allocations

Fixed!
@eolivelli PTAL,thanks!

eolivelli · 2022-09-27T07:18:16Z

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryImpl.java

+ entry.ledgerId = ledgerId;
+ entry.entryId = entryId;
+ entry.skipped = skipped;
+ entry.data = Unpooled.wrappedBuffer(new byte[0]);


do we really need to allocate this ? what about using a constant ?

entry.data = Unpooled.EMPTY_BUFFER ?

Fixed! PTAL,thanks! @eolivelli

eolivelli · 2022-09-27T07:19:34Z

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryImpl.java

@@ -66,6 67,17 @@ public static EntryImpl create(long ledgerId, long entryId, byte[] data) {
 return entry;
 }

+ public static EntryImpl create(long ledgerId, long entryId, boolean skipped) {


what about "createSkippedEntry" ?
otherwise people may want to use this factory method for other purposes

Fixed! PTAL,thanks! @eolivelli

eolivelli

LGTM

lordcheng10 · 2022-09-27T12:00:48Z

@codelipenghui @Technoboy- @Jason918 PTAL,thanks!

Jason918 · 2022-09-27T13:04:51Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java

+ if (entriesToFiltered == null) {
+ entriesToFiltered = new ArrayList<>();
+ }
+ entriesToFiltered.add(entryImpl.getPosition());


Do we need entries.set(i, null); and entry.release();?

OK , I will fix

Fixed, PTAL, thanks! @Jason918

Jason918 · 2022-09-27T13:06:43Z

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpReadEntry.java

+ PositionImpl endPosition = (PositionImpl) nexReadPosition;
+ while (startPosition.compareTo(endPosition) < 0) {
+ skippedEntries.add(EntryImpl.createSkippedEntry(startPosition.ledgerId, startPosition.entryId));
+ startPosition = ledger.getNextValidPosition(startPosition);


It seems, normally, it will just goes from readPosition to nexReadPosition?
Will you miss other entries to be acked?

It seems, normally, it will just goes from readPosition to nexReadPosition?

YES

Will you miss other entries to be acked?

IMO, won't miss other entries to be acked

codelipenghui · 2022-10-08T07:38:49Z

@lordcheng10 Thanks for the clarification. I understand the issue for now.

Can we just ack the messages from the read position(first read failed) to the LAC of the Ledger? For example, if we have a Ledger with 1000 entries. but failed when reading the 501th entry (NonRecoverableLedgerException ). Maybe the Ledger has two fragments. Only the data of the last fragment data is non-recoverable. Then we can just perform an ack operation from 501th to 999th.

So that we don't need to introduce new options to users and this is more intuitive and easier to understand

codelipenghui · 2022-10-08T07:41:24Z

Add a skipped variable in EntryImpl to record whether the entry is automatically skipped;
When pushing entrys to the consumer, automatically ack the skipped entrys;

Sorry, I didn't understand here(From the PR description). If the entry is not able to be read, how can the broker push the entries to the consumer?

lordcheng10 · 2022-10-08T07:46:31Z

Add a skipped variable in EntryImpl to record whether the entry is automatically skipped;
When pushing entrys to the consumer, automatically ack the skipped entrys;

Sorry, I didn't understand here(From the PR description). If the entry is not able to be read, how can the broker push the entries to the consumer?

For unreadable entries, an entry with the skpped flag will be created. When pushing messages, the entries with the skipped flag as true will be filtered out, and then these messages will be acked:

codelipenghui · 2022-10-08T07:57:23Z

For unreadable entries, an entry with the skpped flag will be created. When pushing messages, the entries with the skipped flag as true will be filtered out, and then these messages will be acked

I see. It looks like the read operation will return some fake entries to the caller. Then the caller ack the entries according to the skip flag introduced in the EntryImpl. What is the difference between ack the entries directly when we complete the read operation by checkReadCompletion? In this way, we don't need to introduce a new concept to the EntryImpl, and the caller also has no responsibility to help managed cursor handle internal ack state maintenance issues.

BTW, the Pulsar SQL also uses the cursor API to read data from bookies. If we add skipped field to the EntryImpl, the Pulsar SQL might also need to handle this case.

lordcheng10 · 2022-10-08T08:25:04Z

For unreadable entries, an entry with the skpped flag will be created. When pushing messages, the entries with the skipped flag as true will be filtered out, and then these messages will be acked

I see. It looks like the read operation will return some fake entries to the caller. Then the caller ack the entries according to the skip flag introduced in the EntryImpl. What is the difference between ack the entries directly when we complete the read operation by checkReadCompletion? In this way, we don't need to introduce a new concept to the EntryImpl, and the caller also has no responsibility to help managed cursor handle internal ack state maintenance issues.

BTW, the Pulsar SQL also uses the cursor API to read data from bookies. If we add skipped field to the EntryImpl, the Pulsar SQL might also need to handle this case.

OK, I will fix

mattisonchao · 2022-10-08T08:45:25Z

1 Agree with @codelipenghui concern. individual ack directly is better than introducing new stuff.

lordcheng10 · 2022-10-08T08:52:51Z

1 Agree with @codelipenghui concern. individual ack directly is better than introducing new stuff.

OK , I will fixed!

poorbarcode · 2022-10-08T08:41:23Z

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpReadEntry.java

+ break;
+ }
+ }
+ List<Entry> filteredEntries = cursor.filterReadEntries(skippedEntries);


Can we just drop the data with the instruction cursor.delete(positions)? This saves the memory of entries and is easier to understand.

OK , I will fix

poorbarcode · 2022-10-08T08:53:07Z

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/OpReadEntry.java

+ skippedEntries.add(EntryImpl.createSkippedEntry(startPosition.ledgerId, startPosition.entryId));
+ startPosition = ledger.getNextValidPosition(startPosition);
+ toAckEntryNum ;
+ if (toAckEntryNum > cursor.getConfig().getMaxAckEntryNumForAutoSkipNonRecoverableData()) {


If the state of the cursor is like this:

read position {1:0} individual deleted messages [ {1:0}, {1:50001} ]

After the entry filter, the nexReadPosition will be {1,10000} and filteredEntries will be {1,0}, then maxAckEntryNumForAutoSkipNonRecoverableData could not work perfect. Can we let cursor do the filtering?

github-actions · 2022-11-09T02:15:55Z

The pr had no activity for 30 days, mark with Stale label.

congbobo184 · 2022-11-17T12:13:18Z

@lordcheng10 hi, I move this PR to release/2.9.5, if you have any questions, please ping me. thanks.

lordcheng10 · 2022-12-13T11:56:24Z

OpReadEntry.java

OK

github-actions · 2023-01-13T02:04:10Z

The pr had no activity for 30 days, mark with Stale label.

michaeljmarshall · 2023-06-27T21:56:22Z

As discussed on the mailing list https://lists.apache.org/thread/w4jzk27qhtosgsz7l9bmhf1t7o9mxjhp, there is no plan to release 2.9.6, so I am going to remove the release/2.9.6 label

lordcheng10 changed the title ~~When isAutoSkipNonRecoverableData=true, fix the problem that the mark…~~ [fix][broker] Fix the problem that the markdelete position does not move forward Sep 20, 2022

lordcheng10 changed the title ~~[fix][broker] Fix the problem that the markdelete position does not move forward~~ [fix][broker] Fix the markdelete position does not move forward when isAutoSkipNonRecoverableData=true Sep 20, 2022

lordcheng10 changed the title ~~[fix][broker] Fix the markdelete position does not move forward when isAutoSkipNonRecoverableData=true~~ [fix][broker] Fix the markdelete position does not move forward when isAutoSkipNonRecoverableData=true and individual ack Sep 20, 2022

lordcheng10 mentioned this pull request Sep 20, 2022

[run-tests]Fix the markdelete position does not move forward when isAutoSkipNonRecoverableData=true and individual ack lordcheng10/pulsar#7

Closed

github-actions bot added the doc-not-needed Your PR changes do not impact docs label Sep 20, 2022

lordcheng10 force-pushed the fix_skip_entry_no_ack branch from d1ee2ee to 2f52eb1 Compare September 21, 2022 14:11

lordcheng10 mentioned this pull request Sep 21, 2022

[run-tests] [fix][broker] Fix the markdelete position does not move forward when isAutoSkipNonRecoverableData=true and individual ack lordcheng10/pulsar#14

Closed

lordcheng10 force-pushed the fix_skip_entry_no_ack branch from 2f52eb1 to 253273c Compare September 25, 2022 07:01

lordcheng10 mentioned this pull request Sep 25, 2022

[run-tests] [fix][broker] Fix the markdelete position does not move forward when isAutoSkipNonRecoverableData=true and individual ack lordcheng10/pulsar#19

Open

eolivelli requested changes Sep 25, 2022

View reviewed changes

lordcheng10 requested a review from eolivelli September 25, 2022 13:49

eolivelli requested changes Sep 27, 2022

View reviewed changes

lordcheng10 requested a review from eolivelli September 27, 2022 10:30

eolivelli approved these changes Sep 27, 2022

View reviewed changes

lordcheng10 force-pushed the fix_skip_entry_no_ack branch from 5b0c3a2 to e835c9f Compare September 27, 2022 11:41

Jason918 requested a review from AnonHxy September 27, 2022 12:39

Jason918 assigned lordcheng10 Sep 27, 2022

Jason918 added release/2.9.4 release/2.8.5 release/2.11.1 release/2.10.3 type/bug The PR fixed a bug or issue reported a bug area/broker labels Sep 27, 2022

Jason918 reviewed Sep 27, 2022

View reviewed changes

lordcheng10 requested review from Jason918 and removed request for AnonHxy September 27, 2022 13:43

lordcheng10 requested a review from codelipenghui October 8, 2022 07:28

poorbarcode reviewed Oct 8, 2022

View reviewed changes

github-actions bot added the Stale label Nov 9, 2022

congbobo184 added release/2.9.5 and removed release/2.9.4 labels Nov 17, 2022

github-actions bot removed the Stale label Nov 18, 2022

github-actions bot added the Stale label Jan 13, 2023

liangyepianzhou added release/2.10.4 and removed release/2.10.3 labels Feb 1, 2023

coderzc added release/2.9.6 and removed release/2.9.5 labels Mar 6, 2023

michaeljmarshall removed the release/2.9.6 label Jun 27, 2023

github-actions bot removed the Stale label Jun 28, 2023

Technoboy- added this to the 3.2.0 milestone Jul 31, 2023

Technoboy- modified the milestones: 3.2.0, 3.3.0 Dec 22, 2023

coderzc modified the milestones: 3.3.0, 3.4.0 May 8, 2024

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[fix][broker] Fix the markdelete position does not move forward when isAutoSkipNonRecoverableData=true and individual ack #17751

[fix][broker] Fix the markdelete position does not move forward when isAutoSkipNonRecoverableData=true and individual ack #17751

lordcheng10 commented Sep 20, 2022 •

edited

Loading

lordcheng10 commented Sep 20, 2022

eolivelli Sep 25, 2022

lordcheng10 Sep 25, 2022

lordcheng10 Sep 25, 2022

eolivelli Sep 27, 2022

lordcheng10 Sep 27, 2022 •

edited

Loading

lordcheng10 Sep 27, 2022

eolivelli Sep 27, 2022

lordcheng10 Sep 27, 2022

lordcheng10 Sep 27, 2022

eolivelli left a comment

lordcheng10 commented Sep 27, 2022

Jason918 Sep 27, 2022

lordcheng10 Sep 27, 2022

lordcheng10 Sep 27, 2022

Jason918 Sep 27, 2022

lordcheng10 Sep 27, 2022 •

edited

Loading

codelipenghui commented Oct 8, 2022

codelipenghui commented Oct 8, 2022

lordcheng10 commented Oct 8, 2022 •

edited

Loading

codelipenghui commented Oct 8, 2022

lordcheng10 commented Oct 8, 2022

mattisonchao commented Oct 8, 2022 •

edited

Loading

lordcheng10 commented Oct 8, 2022

poorbarcode Oct 8, 2022

lordcheng10 Oct 9, 2022

poorbarcode Oct 8, 2022

lordcheng10 Oct 9, 2022 •

edited

Loading

github-actions bot commented Nov 9, 2022

congbobo184 commented Nov 17, 2022

lordcheng10 commented Dec 13, 2022

github-actions bot commented Jan 13, 2023

michaeljmarshall commented Jun 27, 2023

	} else if (cursor.config.isAutoSkipNonRecoverableData() && exception instanceof NonRecoverableLedgerException) {
	log.warn("[{}][{}] read failed from ledger at position:{} : {}", cursor.ledger.getName(), cursor.getName(),
	readPosition, exception.getMessage());
	// try to find and move to next valid ledger
	final Position nexReadPosition = cursor.getNextLedgerPosition(readPosition.getLedgerId());
	// fail callback if it couldn't find next valid ledger
	if (nexReadPosition == null) {
	callback.readEntriesFailed(exception, ctx);
	cursor.ledger.mbean.recordReadEntriesError();
	recycle();
	return;
	}
	updateReadPosition(nexReadPosition);
	checkReadCompletion();
	} else {

[fix][broker] Fix the markdelete position does not move forward when isAutoSkipNonRecoverableData=true and individual ack #17751

Are you sure you want to change the base?

[fix][broker] Fix the markdelete position does not move forward when isAutoSkipNonRecoverableData=true and individual ack #17751

Conversation

lordcheng10 commented Sep 20, 2022 • edited Loading

Motivation

Modifications

Documentation

Matching PR in forked repository

lordcheng10 commented Sep 20, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lordcheng10 Sep 27, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

eolivelli left a comment

Choose a reason for hiding this comment

lordcheng10 commented Sep 27, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lordcheng10 Sep 27, 2022 • edited Loading

Choose a reason for hiding this comment

codelipenghui commented Oct 8, 2022

codelipenghui commented Oct 8, 2022

lordcheng10 commented Oct 8, 2022 • edited Loading

codelipenghui commented Oct 8, 2022

lordcheng10 commented Oct 8, 2022

mattisonchao commented Oct 8, 2022 • edited Loading

lordcheng10 commented Oct 8, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lordcheng10 Oct 9, 2022 • edited Loading

Choose a reason for hiding this comment

github-actions bot commented Nov 9, 2022

congbobo184 commented Nov 17, 2022

lordcheng10 commented Dec 13, 2022

github-actions bot commented Jan 13, 2023

michaeljmarshall commented Jun 27, 2023

lordcheng10 commented Sep 20, 2022 •

edited

Loading

lordcheng10 Sep 27, 2022 •

edited

Loading

lordcheng10 Sep 27, 2022 •

edited

Loading

lordcheng10 commented Oct 8, 2022 •

edited

Loading

mattisonchao commented Oct 8, 2022 •

edited

Loading

lordcheng10 Oct 9, 2022 •

edited

Loading