[Transaction] Offload transaction data #12449

liangyepianzhou · 2021-10-21T06:42:55Z

Master Issue: #12482
PIP : streamnative/community#6

Motivation

Redundant information does not need to be stored for a long time. The messages that have been aborted in a transaction and the Txn Marks that have been read are all redundant information

Modifications

Only Offload the length and entryId of entries that do not need to be Offloaded.

Modified the offload method in ledgerOffload interface, and added an OffloadFilter type parameter.
OffloadFilter is a newly added interface, responsible for judging from the perspective of transaction,
whether the ledger can be uninstalled, and the entry needs to be offloaded and so on. The main implementation is OffloadFilterImp and OffloadFilterDisabled

Verifying this change

Add a unit test in pulsar-broker.

Make sure that the change passes the CI checks.

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

If yes was chosen, please highlight the changes

Dependencies (does it add or upgrade a dependency): (yes / no)
The public API: (yes / no)
The schema: (yes / no / don't know)
The default values of configurations: (yes / no)
The wire protocol: (yes / no)
The rest endpoints: (yes / no)
The admin cli options: (yes / no)
Anything that affects deployment: (yes / no / don't know)

Documentation

Check the box below and label this PR (if you have committer privilege).

Need to update docs?

no-need-doc
This is a perfection of existing functions

The implementation class of mock is not modified

…d, and a part of the entryId is bigger than nextExcpectedId

…r without processing. Give up constantly takingSnapshot.

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

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

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

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

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

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

congbobo184 · 2021-10-21T12:40:58Z

...ava/org/apache/bookkeeper/mledger/offload/filesystem/impl/FileStoreBackedReadHandleImpl.java

                reader.seek(key);
-                while (entriesToRead > 0) {
-                    reader.next(key, value);
+                reader.next(key, value);


why not use while do? if reader.next(key, value) return false, value.getLength() will throw NPE

value.getLength() will be 0. And the entries will get. a entry which the length is 0.
And then the exception will not be thrown when LedgerEntriesImpl.create(entries).

.../java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreManagedLedgerOffloader.java

congbobo184 · 2021-10-21T14:44:41Z

...in/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreBackedReadHandleImpl.java

-                        continue;
+                    long entryId;
+                    int length;
+                    try {


dont use try catch, this code have been created in try catch line 110

The two try catch here are different. The try catch here is to obtain the EOF exception caused by the last few entries in the ledger that have been fittered, and exit the reading loop when the EOF exception is obtained.

congbobo184

left some comments

congbobo184 · 2021-10-22T01:44:16Z

...in/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreBackedReadHandleImpl.java


                while (entriesToRead > 0) {
                    if (state == State.Closed) {
                        log.warn("Reading a closed read handler. Ledger ID: {}, Read range: {}-{}", ledgerId, firstEntry, lastEntry);


we should delete entriesToRead right? because we may not read the entriesToRead number of entries

If there is no transaction messages, this is still useful. Can we consider keeping it？

congbobo184 · 2021-10-22T01:44:42Z

...in/java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreBackedReadHandleImpl.java

+                if (entries.isEmpty()) {
+                    ByteBuf buf = PooledByteBufAllocator.DEFAULT.buffer(0, 0);
+                    entries.add(LedgerEntryImpl.create(ledgerId, 0, 0, buf));
+                }


why we should add this logic?

promise.complete(LedgerEntriesImpl.create(entries));
The method of LedgerEntriesImpl.create(entries) will check if entries are empty.

congbobo184 · 2021-10-22T01:45:10Z

.../java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreBackedReadHandleImplV2.java

-                                            .getIndexEntryForEntry(groupedReader.ledgerId, nextExpectedId)
-                                            .getDataOffset());
-                            continue;
+


same as BlobStoreBackedReadHandleImpl.java

liangyepianzhou · 2021-11-01T05:49:35Z

/pulsarbot run-failure-checks

congbobo184 · 2021-11-03T06:17:04Z

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

-        if (opReadEntry.readPosition.compareTo(opReadEntry.maxPosition) > 0) {
+        //entries that has been offloaded do not need to be restricted by TransactionBuffer`maxReadPosition
+        LedgerInfo info = ledgers.get(ledger.getId());
+        if (opReadEntry.readPosition.compareTo(opReadEntry.maxPosition) > 0


we only can offload small than maxPosition? why we should add this logic?

We don’t know if the messages after MaxReadPosition are aborted.

Why the messages are aborted or not is related to offloading?

congbobo184 · 2021-11-03T06:18:27Z

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

        // can read max position entryId
        if (ledger.getId() == opReadEntry.maxPosition.getLedgerId()) {
-            lastEntryInLedger = min(opReadEntry.maxPosition.getEntryId(), lastEntryInLedger);
+            if(!info.getOffloadContext().getComplete()){


same as above

If no transaction message has been sent, then MaxReadPosition does not need to be considered when recovering from tiered storage.

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

.../java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreManagedLedgerOffloader.java

congbobo184 · 2021-11-03T08:18:31Z

.../java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreBackedReadHandleImplV2.java

+                            }
+                            entryId = groupedReader.dataStream.readLong();
+                        } catch (EOFException ioException) {
+                            //Some entries in this ledger may be filtered. If the last entry in this ledger was filtered,


add a test for it when produce EOFException

congbobo184 · 2021-11-03T08:20:52Z

.../java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreBackedReadHandleImplV2.java

+                    groupedReader.inputStream
+                            .seek(groupedReader.index
+                                    .getIndexEntryForEntry(groupedReader.ledgerId, nextExpectedId)
+                                    .getDataOffset());


Annotate this seems to have no effect on test

congbobo184 · 2021-11-03T08:24:08Z

.../java/org/apache/bookkeeper/mledger/offload/jcloud/impl/BlobStoreBackedReadHandleImplV2.java

                    }
+                    if (entries.isEmpty()) {
+                        ByteBuf buf = PooledByteBufAllocator.DEFAULT.buffer(0, 0);
+                        entries.add(LedgerEntryImpl.create(ledgerId, 0, 0, buf));


we can't do this, because LedgerEntryImpl.create(ledgerId, 0, 0, buf) will be sent to consumer but it don't have any data

congbobo184 · 2021-11-03T08:35:32Z

...ava/org/apache/bookkeeper/mledger/offload/filesystem/impl/FileStoreBackedReadHandleImpl.java

-                    }
-            }
+                    entryId = key.get();
+                    ByteBuf buf = PooledByteBufAllocator.DEFAULT.buffer(length, length);


we can't return length is 0 entry, same as above BlockAwareSegmentInputStream.java

congbobo184

LGTM! left some comment

congbobo184 · 2021-11-17T07:19:13Z

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

     * @param uid unique id to identity this offload attempt
     * @param extraMetadata metadata to be stored with the offloaded ledger for informational
     *                      purposes
+     * @param offloadFilter  Filter out unnecessary entries


Filter lower case

congbobo184 · 2021-11-17T07:46:30Z

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

        }

        // can read max position entryId
        if (ledger.getId() == opReadEntry.maxPosition.getLedgerId()) {


ledger.getId() == opReadEntry.maxPosition.getLedgerId() && !info.getOffloadContext().getComplete()

Why need this check?

congbobo184 · 2021-11-17T08:19:50Z

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

-
+                                if (persistentTopic.getBrokerService().getPulsar().getConfig()
+                                        .isTransactionCoordinatorEnabled()){
+                                    if (ledger instanceof ManagedLedgerImpl) {


add a interface to ManagedLedger is better, right?

congbobo184 · 2021-11-17T10:12:28Z

pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/OffloadTxnDataTest.java

+    private static final String NAMESPACE1 = TENANT + "/ns1";
+    private static final int NUM_BROKERS = 1;
+    private static final int NUM_PARTITIONS = 1;
+    private Properties properties = new Properties();


seem not to use

congbobo184 · 2021-11-17T10:19:25Z

pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/OffloadTxnDataTest.java

+        this.setBrokerCount(NUM_BROKERS);
+        this.internalSetup();
+        String[] brokerServiceUrlArr = getPulsarServiceList().get(0).getBrokerServiceUrl().split(":");
+        String webServicePort = brokerServiceUrlArr[brokerServiceUrlArr.length - 1];
+        admin.clusters().createCluster(CLUSTER_NAME, ClusterData.builder()
+                .serviceUrl("http://localhost:" + webServicePort).build());
+        admin.tenants().createTenant(TENANT,
+                new TenantInfoImpl(Sets.newHashSet("appid1"), Sets.newHashSet(CLUSTER_NAME)));
+        admin.namespaces().createNamespace(NAMESPACE1);
+
+        admin.tenants().createTenant(NamespaceName.SYSTEM_NAMESPACE.getTenant(),
+                new TenantInfoImpl(Sets.newHashSet("appid1"), Sets.newHashSet(CLUSTER_NAME)));
+        admin.namespaces().createNamespace(NamespaceName.SYSTEM_NAMESPACE.toString());
+        admin.topics().createPartitionedTopic(TopicName.TRANSACTION_COORDINATOR_ASSIGN.toString(), NUM_PARTITIONS);
+        pulsarClient.close();
+        pulsarClient = PulsarClient.builder()
+                .serviceUrl(getPulsarServiceList().get(0).getBrokerServiceUrl())
+                .statsInterval(0, TimeUnit.SECONDS)
+                .enableTransaction(true)
+                .build();
+        // wait tc init success to ready state
+        waitForCoordinatorToBeAvailable(NUM_PARTITIONS);


same to TransactionTestBase.setUpBase, could you please code like MockedPulsarServiceBaseTest.internalSetup?

congbobo184 · 2021-11-17T10:20:42Z

pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/OffloadTxnDataTest.java

+        fileSystemManagedLedgerOffloader.getOffloadPolicies().setManagedLedgerOffloadThresholdInBytes(1L);
+        setLedgerOffloader(fileSystemManagedLedgerOffloader);
+        setMaxEntriesPerLedger(4);
+        setProperties(properties);


seem not to use

gaoran10

Nice work! Left some comments.

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

gaoran10 · 2021-11-17T10:46:22Z

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

-        if (opReadEntry.readPosition.compareTo(opReadEntry.maxPosition) > 0) {
+        //entries that has been offloaded do not need to be restricted by TransactionBuffer`maxReadPosition
+        LedgerInfo info = ledgers.get(ledger.getId());
+        if (opReadEntry.readPosition.compareTo(opReadEntry.maxPosition) > 0


Why the messages are aborted or not is related to offloading?

gaoran10 · 2021-11-17T10:51:49Z

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

        }

        // can read max position entryId
        if (ledger.getId() == opReadEntry.maxPosition.getLedgerId()) {


Why need this check?

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

gaoran10 · 2021-11-17T10:56:46Z

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


    private ManagedLedgerInterceptor managedLedgerInterceptor;

+    private OffloadFilter offloadFilter;


Maybe we could use a list here? It seems that the offload filter may have different implementations.

Why don't we add methods in it？

gaoran10 · 2021-11-17T11:08:16Z

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

            if (entry == null) {
                continue;
+            } else if (entry.getLength() == 0) {
+                subscription.acknowledgeMessage(Collections.singletonList(entry.getPosition()), AckType.Individual,


Please add a description here.

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

pulsar-broker/pom.xml

github-actions · 2022-02-27T01:55:44Z

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

liangyepianzhou added 18 commits September 28, 2021 20:36

Optimize takeSnapshot which has contains in a pre PR

bb306d4

Implement offload Components

4e44f12

Set offloadFilter to LedgerOffloader.

1d4626c

The implementation class of mock is not modified

Filter unnecessary transaction entry when offload

eedd8d1

Avoid the infinite loop caused by the transaction entry being filtere…

195b62e

…d, and a part of the entryId is bigger than nextExcpectedId

Implement interface methods for mock classes

95886b0

No need to check the status of TransactionBuffer

90e43af

Messages that have been offloaded can be sent directly to the consume…

4534a37

…r without processing. Give up constantly takingSnapshot.

review details self

fc30a1b

Read test

e2ca171

checkstyle

12ba195

merge

70e78de

Use another implementation

1123e15

Details

a0f0a49

update

2fee22a

update

c767eb0

update

937f544

update

37232a9

eolivelli assigned liangyepianzhou Oct 21, 2021

eolivelli added the doc-not-needed Your PR changes do not impact docs label Oct 21, 2021

update

8cf65f9

congbobo184 requested changes Oct 21, 2021

View reviewed changes

congbobo184 requested changes Oct 22, 2021

View reviewed changes

liangyepianzhou added 7 commits October 22, 2021 11:03

update

d530005

update

ab80354

Fix tests

2d520e0

Optimize code

8d35424

merge apache/master

9e6634c

Fix tests

c157b34

Fix dependency

441eea2

liangyepianzhou requested a review from congbobo184 November 1, 2021 07:13

congbobo184 requested changes Nov 3, 2021

View reviewed changes

liangyepianzhou added 8 commits November 3, 2021 18:22

Optimize

8d7e8b1

Optimize

9cf8087

Merge

0c3abb1

Optimize

b2c9844

Optimize

73a39a7

Optimize

bc12011

Fix and remove logs

ac94c59

Optimize ReadHandle

c9b36fb

liangyepianzhou mentioned this pull request Nov 11, 2021

The problem of two exception handling #12744

Merged

2 tasks

liangyepianzhou added 3 commits November 16, 2021 18:42

Pass a the empty databuf to the entry that does not require offload

b60edac

Remove a NonRecoverableLedgerException

6cb30a4

update

3dddb51

congbobo184 reviewed Nov 17, 2021

View reviewed changes

gaoran10 reviewed Nov 17, 2021

View reviewed changes

liangyepianzhou added 2 commits November 17, 2021 21:45

update

8037398

move test

7ef9d6f

hangc0276 requested review from codelipenghui, hangc0276 and zymap November 19, 2021 10:25

optimize pom

a825908

codelipenghui added this to the 2.11.0 milestone Jan 21, 2022

github-actions bot added the lifecycle/stale label Feb 27, 2022

codelipenghui modified the milestones: 2.11.0, 2.12.0 Jul 26, 2022

liangyepianzhou closed this Nov 18, 2022

dao-jun mentioned this pull request May 19, 2024

[Transaction]Offload Transaction Data #12482

Closed


		private ManagedLedgerInterceptor managedLedgerInterceptor;

		private OffloadFilter offloadFilter;

[Transaction] Offload transaction data #12449

[Transaction] Offload transaction data #12449

Uh oh!

Conversation

liangyepianzhou commented Oct 21, 2021 • 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

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

congbobo184 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

liangyepianzhou commented Nov 1, 2021

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

congbobo184 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

congbobo184 Nov 17, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

liangyepianzhou commented Oct 21, 2021 •

edited

Loading

congbobo184 Nov 17, 2021 •

edited

Loading