I've opened a PR which adds an integration test <https://github.com/apache/kafka/pull/17668/files#diff-a60b518846fc0f770164d55b6d7cd31e03a002514377578c80c6e22cc120af40R88> to show the impact of this change/KIP. PTAL.
-- Kamal On Fri, Nov 1, 2024 at 3:31 PM Kamal Chandraprakash < kamal.chandraprak...@gmail.com> wrote: > Thanks for the review! > > > 5 > Updated the KIP to include the `isEmpty` method in the transaction index > > > 6 > You're right. The offset parameter will be equal to the > next-segment-to-consider base offset. > But, the API introduced in RLMM is for *one* epoch. The next epoch > start-offset may not be > the base-offset. You can refer to this line > <https://github.com/apache/kafka/pull/17659/files#diff-380e4d8859ea9148f21794c09039425c82d9012a392c2dbbe1ce2ec8677a1970R1857> > in the draft PR. > > > On Fri, Nov 1, 2024 at 3:16 PM Divij Vaidya <divijvaidy...@gmail.com> > wrote: > >> Last few things - >> >> # 5 >> About setting the TrxIndexEmpty field, could we introduce an isEmpty() >> function in TransactionIndex which has the following implementation: >> >> public boolean isEmpty() { >> return !iterable().iterator().hasNext(); >> } >> >> >> The advantages of this approach is: >> 1. It works for both cases when the file is not present and also when the >> file is present but is empty. >> 2. It prevents leaking the underlying implementation of TransactionIndex >> outside via the file() method. I think that making file() as public is an >> implementation leak (for example, what is the trx indx is not file >> based!). >> >> >> #6 >> In the documentation for nextSegmentWithTxnIndex, the offset parameter >> should be equal to the next-segment-to-consider's base offset, no? >> I assume that we will add a new fetch here with nextSegmentBaseOffset >> >> https://github.com/apache/kafka/blob/346fdbafc539bc48bb66eedae89a15e240007fd9/core/src/main/java/kafka/log/remote/RemoteLogManager.java#L1801 >> . Is there a case where the parameter "offset" will not be equal to the >> baseOffset of a segment? >> >> -- >> Divij Vaidya >> >> >> >> On Fri, Nov 1, 2024 at 10:26 AM Kamal Chandraprakash < >> kamal.chandraprak...@gmail.com> wrote: >> >> > Hi Divij, >> > >> > Thanks for the detailed review! >> > >> > > 1, 2, 3, 4 >> > Updated the KIP-1058 >> > < >> > >> https://cwiki.apache.org/confluence/display/KAFKA/KIP-1058%3A+Txn+consumer+exerts+pressure+on+remote+storage+when+collecting+aborted+transactions >> > > >> > with the feedback received and also opened a draft PR for #17659 >> > <https://github.com/apache/kafka/pull/17659/files> reference. >> > PTAL. >> > >> > > 5. How are we determining the value of the TrxIndexEmpty field on >> segment >> > rotation? >> > Transaction index file is optional, the file does not exists when there >> are >> > no aborted txn entries for a >> > segment, we will be using the file null check. Also, updated it in the >> KIP. >> > >> > Thanks, >> > Kamal >> > >> > >> > On Tue, Oct 29, 2024 at 8:47 PM Divij Vaidya <divijvaidy...@gmail.com> >> > wrote: >> > >> > > A few more points to discuss (please add to the KIP as well) >> > > >> > > 5. How are we determining the value of the TrxIndexEmpty field on >> segment >> > > rotation? >> > > >> > > One option is to do a boolean txnIdxEmpty = >> > > segment.txnIndex().allAbortedTxns().isEmpty() but this will have an >> > > overhead of reading the contents of the file and storing them in >> memory, >> > > when we have a non-empty index. >> > > The other option (preferred) is to add an isEmpty() public method to >> the >> > > TransactionIndex and perform a segment.txnIndex().isEmpty() check >> which >> > > will internally use Files.size() java API. >> > > >> > > On Tue, Oct 29, 2024 at 1:21 PM Divij Vaidya <divijvaidy...@gmail.com >> > >> > > wrote: >> > > >> > > > Let's get the ball rolling (again) on this one. >> > > > >> > > > Kamal, could you please add the following to the KIP: >> > > > 1. the API as discussed above. Please add the failure modes for this >> > API >> > > > as well such as the exceptions thrown and a recommendation on how a >> > > caller >> > > > is expected to handle those. I am assuming that the three parameters >> > for >> > > > this API will be topicPartition, epoch and offset. >> > > > 2. implementation details for Topic based RLMM. I am assuming that >> the >> > > > plugin will default the field to false if this field is absent >> (case of >> > > old >> > > > metadata). >> > > > 3. In the test plan section, additionally, we need to assert that we >> > > don't >> > > > read metadata for all segments (i.e. it is not a linear search) from >> > the >> > > > Topic based RLMM. >> > > > 4. in the compatibility section, please document how the existing >> > > clusters >> > > > with Tiered Storage metadata will work during/after a rolling >> upgrade >> > to >> > > a >> > > > version which contains this new change. >> > > > >> > > > -- >> > > > Divij Vaidya >> > > > >> > > > >> > > > >> > > > On Fri, Oct 11, 2024 at 12:26 PM Kamal Chandraprakash < >> > > > kamal.chandraprak...@gmail.com> wrote: >> > > > >> > > >> Bump for review. >> > > >> >> > > >> If the additional proposal looks good, I'll append them to the KIP. >> > > PTAL. >> > > >> >> > > >> New API in RLMM#nextRemoteLogSegmentMetadataWithTxnIndex >> > > >> >> > > >> -- >> > > >> Kamal >> > > >> >> > > >> On Sun, Oct 6, 2024 at 7:20 PM Kamal Chandraprakash < >> > > >> kamal.chandraprak...@gmail.com> wrote: >> > > >> >> > > >> > Hi Christo, >> > > >> > >> > > >> > Thanks for the review! >> > > >> > >> > > >> > Adding the new API `nextRemoteLogSegmentMetadataWithTxnIndex` in >> > RLMM >> > > >> > helps to >> > > >> > reduce the complexity of linear search. With this API, we have >> to: >> > > >> > >> > > >> > 1. Maintain one more skip-list [1] for each of the epochs in the >> > > >> partition >> > > >> > in RLMM that might >> > > >> > increase the memory usage of TopicBased RLMM implementation. >> > > >> > 1a) The skip-list will be empty when there are no aborted txn >> > > >> entries >> > > >> > for a partition/epoch which is the predominant case. >> > > >> > 1b) The skip-list will act as a duplicate when *most* of the >> > > >> segments >> > > >> > have aborted txn entries, assuming aborted txn are quite low, >> this >> > > >> should >> > > >> > be fine. >> > > >> > 2. Change the logic to retrieve the aborted txns (we have to >> query >> > the >> > > >> > nextRLSMWithTxnIndex >> > > >> > for each of the leader-epoch). >> > > >> > 3. Logic divergence from how we retrieve the aborted txn entries >> > > >> compared >> > > >> > to the local-log. >> > > >> > >> > > >> > The approach looks good to me. If everyone is aligned, then we >> can >> > > >> proceed >> > > >> > to add this API to RLMM. >> > > >> > >> > > >> > Another option I was thinking of is to capture the >> > > `lastStableOffsetLag` >> > > >> > [2] while rotating the segment. >> > > >> > But, that is a bigger change we can take later. >> > > >> > >> > > >> > [1]: >> > > >> > >> > > >> >> > > >> > >> https://sourcegraph.com/github.com/apache/kafka/-/blob/storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/RemoteLogLeaderEpochState.java?L43 >> > > >> > [2]: >> > > >> > >> > > >> >> > > >> > >> https://sourcegraph.com/github.com/apache/kafka/-/blob/core/src/main/scala/kafka/log/UnifiedLog.scala?L432 >> > > >> > >> > > >> > >> > > >> > Thanks, >> > > >> > Kamal >> > > >> > >> > > >> > On Fri, Oct 4, 2024 at 4:21 PM Christo Lolov < >> > christolo...@gmail.com> >> > > >> > wrote: >> > > >> > >> > > >> >> Heya, >> > > >> >> >> > > >> >> Apologies for the delay. I have been thinking about this problem >> > > >> recently >> > > >> >> as well and while I believe storing a boolean in the metadata is >> > > good, >> > > >> I >> > > >> >> think we can do better by introducing a new method to the RLMM >> > along >> > > >> the >> > > >> >> lines of >> > > >> >> >> > > >> >> Optional<RemoteLogSegmentMetadata> >> > > >> >> nextRemoteLogSegmentMetadataWithTxnIndex(TopicIdPartition >> > > >> >> topicIdPartition, >> > > >> >> int epochForOffset, long offset) throws RemoteStorageException >> > > >> >> >> > > >> >> This will help plugin implementers to build optimisations such >> as >> > > skip >> > > >> >> lists which will give them the next segment quicker than a >> linear >> > > >> search. >> > > >> >> >> > > >> >> I am keen to hear your thoughts! >> > > >> >> >> > > >> >> Best, >> > > >> >> Christo >> > > >> >> >> > > >> >> On Fri, 4 Oct 2024 at 10:48, Kamal Chandraprakash < >> > > >> >> kamal.chandraprak...@gmail.com> wrote: >> > > >> >> >> > > >> >> > Hi Luke, >> > > >> >> > >> > > >> >> > Thanks for the review! >> > > >> >> > >> > > >> >> > > Do you think it is helpful if we store the "least abort >> start >> > > >> offset >> > > >> >> in >> > > >> >> > the >> > > >> >> > segment", and -1 means no txnIndex. So that we can have a way >> to >> > > know >> > > >> >> if we >> > > >> >> > need to fetch this txn index or not. >> > > >> >> > >> > > >> >> > 1. No, this change won't have an effect. To find the >> upper-bound >> > > >> offset >> > > >> >> > [1], we have to >> > > >> >> > fetch that segment's offset index file. The >> RemoteIndexCache >> > > [2] >> > > >> >> > fetches all the 3 >> > > >> >> > index files together and caches them for subsequent use, >> so >> > > this >> > > >> >> > improvement >> > > >> >> > won't have an effect as the current segment txn index gets >> > > >> >> downloaded >> > > >> >> > anyway. >> > > >> >> > >> > > >> >> > 2. The reason for choosing boolean is to make the change >> backward >> > > >> >> > compatible. >> > > >> >> > There can be existing RLM events for the uploaded >> segments. >> > > The >> > > >> >> > default >> > > >> >> > value of `txnIdxEmpty` is false so the *old* RLM events >> are >> > > >> >> assumed to >> > > >> >> > contain >> > > >> >> > the txn index files and those files are downloaded if >> they >> > > >> exist. >> > > >> >> > >> > > >> >> > [1]: >> > > >> >> > >> > > >> >> > >> > > >> >> >> > > >> >> > > >> > >> https://sourcegraph.com/github.com/apache/kafka@trunk/-/blob/core/src/main/java/kafka/log/remote/RemoteLogManager.java?L1732 >> > > >> >> > [2]: >> > > >> >> > >> > > >> >> > >> > > >> >> >> > > >> >> > > >> > >> https://sourcegraph.com/github.com/apache/kafka@trunk/-/blob/storage/src/main/java/org/apache/kafka/storage/internals/log/RemoteIndexCache.java?L383 >> > > >> >> > >> > > >> >> > Thanks, >> > > >> >> > Kamal >> > > >> >> > >> > > >> >> > On Thu, Oct 3, 2024 at 3:11 PM Luke Chen <show...@gmail.com> >> > > wrote: >> > > >> >> > >> > > >> >> > > Hi Kamal, >> > > >> >> > > >> > > >> >> > > Sorry for the late review. >> > > >> >> > > Thanks for the KIP, this will improve the transaction >> reading >> > for >> > > >> >> remote >> > > >> >> > > storage. >> > > >> >> > > Overall LGTM, just one minor thought: >> > > >> >> > > >> > > >> >> > > Currently, we only store the `TxnIndexEmpty` bool value in >> the >> > > >> segment >> > > >> >> > > metadata. >> > > >> >> > > Do you think it is helpful if we store the "least abort >> start >> > > >> offset >> > > >> >> in >> > > >> >> > the >> > > >> >> > > segment", and -1 means no txnIndex. So that we can have a >> way >> > to >> > > >> know >> > > >> >> if >> > > >> >> > we >> > > >> >> > > need to fetch this txn index or not. >> > > >> >> > > >> > > >> >> > > Thanks. >> > > >> >> > > Luke >> > > >> >> > > >> > > >> >> > > On Mon, Sep 9, 2024 at 3:26 PM Kamal Chandraprakash < >> > > >> >> > > kamal.chandraprak...@gmail.com> wrote: >> > > >> >> > > >> > > >> >> > > > Hi all, >> > > >> >> > > > >> > > >> >> > > > If there are no more comments, I'll start a voting thread >> > soon. >> > > >> >> > > > >> > > >> >> > > > Thanks, >> > > >> >> > > > Kamal >> > > >> >> > > > >> > > >> >> > > > On Fri, Sep 6, 2024 at 7:28 PM Kamal Chandraprakash < >> > > >> >> > > > kamal.chandraprak...@gmail.com> wrote: >> > > >> >> > > > >> > > >> >> > > > > Bumping this thread again for review! >> > > >> >> > > > > >> > > >> >> > > > > Reduced the scope of the proposal to minimum. We will be >> > > adding >> > > >> >> only >> > > >> >> > > one >> > > >> >> > > > > field (txnIdxEmpty) to the >> > > >> >> > > > > RemoteLogSegmentMetadata event which is backward >> > compatible. >> > > >> PTAL. >> > > >> >> > > > > >> > > >> >> > > > > Thanks, >> > > >> >> > > > > Kamal >> > > >> >> > > > > >> > > >> >> > > > > >> > > >> >> > > > > On Tue, Aug 13, 2024 at 6:33 PM Kamal Chandraprakash < >> > > >> >> > > > > kamal.chandraprak...@gmail.com> wrote: >> > > >> >> > > > > >> > > >> >> > > > >> Bumping this thread for KIP review! >> > > >> >> > > > >> >> > > >> >> > > > >> We can go for the simplest solution that is proposed in >> > this >> > > >> KIP >> > > >> >> and >> > > >> >> > > > >> it can be improved in the subsequent iteration. PTAL. >> > > >> >> > > > >> >> > > >> >> > > > >> Thanks, >> > > >> >> > > > >> Kamal >> > > >> >> > > > >> >> > > >> >> > > > >> On Fri, Aug 2, 2024 at 11:42 AM Kamal Chandraprakash < >> > > >> >> > > > >> kamal.chandraprak...@gmail.com> wrote: >> > > >> >> > > > >> >> > > >> >> > > > >>> Hi Divij, >> > > >> >> > > > >>> >> > > >> >> > > > >>> Thanks for the review! And, sorry for the late reply. >> > > >> >> > > > >>> >> > > >> >> > > > >>> From the UnifiedLog.scala >> > > >> >> > > > >>> < >> > > >> >> > > > >> > > >> >> > > >> > > >> >> > >> > > >> >> >> > > >> >> > > >> > >> https://sourcegraph.com/github.com/apache/kafka@trunk/-/blob/core/src/main/scala/kafka/log/UnifiedLog.scala?L421-427 >> > > >> >> > > > > >> > > >> >> > > > >>> doc: >> > > >> >> > > > >>> >> > > >> >> > > > >>> """ >> > > >> >> > > > >>> The last stable offset (LSO) is defined as the first >> > offset >> > > >> such >> > > >> >> > that >> > > >> >> > > > >>> all lower offsets have been "decided." >> > > >> >> > > > >>> * Non-transactional messages are considered decided >> > > >> >> immediately, >> > > >> >> > > but >> > > >> >> > > > >>> transactional messages are only decided when >> > > >> >> > > > >>> * the corresponding COMMIT or ABORT marker is >> written. >> > > >> This >> > > >> >> > > implies >> > > >> >> > > > >>> that the last stable offset will be equal >> > > >> >> > > > >>> * to the high watermark if there are no >> transactional >> > > >> >> messages >> > > >> >> > in >> > > >> >> > > > the >> > > >> >> > > > >>> log. Note also that the LSO cannot advance >> > > >> >> > > > >>> * beyond the high watermark. >> > > >> >> > > > >>> """ >> > > >> >> > > > >>> While rolling the active segment to passive, if LSO >> > equals >> > > to >> > > >> >> HW, >> > > >> >> > > then >> > > >> >> > > > >>> all the messages in that segment are >> > > >> >> > > > >>> decided and we can store the `lastStableOffsetLag` as >> an >> > > >> >> attribute >> > > >> >> > in >> > > >> >> > > > >>> the rolled segment. We can then propagate >> > > >> >> > > > >>> the `lastStableOffsetLag` information in the >> > > >> RemoteLogMetadata >> > > >> >> > > events. >> > > >> >> > > > >>> >> > > >> >> > > > >>> While reading the remote log segment, if the >> > > >> >> `lastStableOffsetLag` >> > > >> >> > is >> > > >> >> > > > 0, >> > > >> >> > > > >>> then there is no need to traverse to >> > > >> >> > > > >>> the subsequent segments for aborted transactions which >> > > covers >> > > >> >> the >> > > >> >> > > case >> > > >> >> > > > >>> for the dominant case where the >> > > >> >> > > > >>> partition had no transactions at all. >> > > >> >> > > > >>> >> > > >> >> > > > >>> With Log compaction, the shrinked segments might get >> > > merged. >> > > >> One >> > > >> >> > > option >> > > >> >> > > > >>> is to take the max of `lastStableOffsetLag` >> > > >> >> > > > >>> and store it in the new LogSegment. Since, the tiered >> > > storage >> > > >> >> does >> > > >> >> > > not >> > > >> >> > > > >>> support compacted topics / historical compacted >> > > >> >> > > > >>> topics, we can omit this case. >> > > >> >> > > > >>> >> > > >> >> > > > >>> If this approach looks good, I can update the KIP with >> > the >> > > >> >> details. >> > > >> >> > > > >>> >> > > >> >> > > > >>> -- >> > > >> >> > > > >>> Kamal >> > > >> >> > > > >>> >> > > >> >> > > > >>> >> > > >> >> > > > >>> >> > > >> >> > > > >>> >> > > >> >> > > > >>> On Tue, Jun 25, 2024 at 4:24 PM Divij Vaidya < >> > > >> >> > > divijvaidy...@gmail.com> >> > > >> >> > > > >>> wrote: >> > > >> >> > > > >>> >> > > >> >> > > > >>>> Hi Kamal >> > > >> >> > > > >>>> >> > > >> >> > > > >>>> Thanks for the bump. I have been thinking about this >> > > >> passively >> > > >> >> for >> > > >> >> > > the >> > > >> >> > > > >>>> past >> > > >> >> > > > >>>> few days. >> > > >> >> > > > >>>> >> > > >> >> > > > >>>> The simplest solution is to store a state at segment >> > level >> > > >> >> > metadata. >> > > >> >> > > > The >> > > >> >> > > > >>>> state should specify whether the trx index is empty >> or >> > > not. >> > > >> It >> > > >> >> > would >> > > >> >> > > > be >> > > >> >> > > > >>>> populated during segment archival. We would then >> iterate >> > > >> over >> > > >> >> the >> > > >> >> > > > >>>> metadata >> > > >> >> > > > >>>> for future segments without having to make a remote >> call >> > > to >> > > >> >> > download >> > > >> >> > > > the >> > > >> >> > > > >>>> trx index itself. >> > > >> >> > > > >>>> >> > > >> >> > > > >>>> The other solution for storing state at a partition >> > level >> > > >> >> wouldn't >> > > >> >> > > > >>>> work, as >> > > >> >> > > > >>>> you mentioned, because we will have to change the >> state >> > on >> > > >> >> every >> > > >> >> > > > >>>> mutation >> > > >> >> > > > >>>> to the log i.e. at expiration of segments and append. >> > > >> >> > > > >>>> >> > > >> >> > > > >>>> I have been thinking whether we can do something >> better >> > > than >> > > >> >> the >> > > >> >> > > > simple >> > > >> >> > > > >>>> solution, hence the delay in replying. Let me tell >> you >> > my >> > > >> half >> > > >> >> > baked >> > > >> >> > > > >>>> train >> > > >> >> > > > >>>> of thoughts, perhaps, you can explore this as well. I >> > have >> > > >> been >> > > >> >> > > > thinking >> > > >> >> > > > >>>> about using LSO (last stable offset) to handle the >> case >> > > when >> > > >> >> the >> > > >> >> > > > >>>> partition >> > > >> >> > > > >>>> never had any transactions. For a partition which >> never >> > > had >> > > >> any >> > > >> >> > > > >>>> transaction, I would assume that the LSO is never >> > > >> initialized >> > > >> >> (or >> > > >> >> > is >> > > >> >> > > > >>>> equal >> > > >> >> > > > >>>> to log start offset)? Or is it equal to HW in that >> case? >> > > >> This >> > > >> >> is >> > > >> >> > > > >>>> something >> > > >> >> > > > >>>> that I am yet to verify. If this idea works, then we >> > would >> > > >> not >> > > >> >> > have >> > > >> >> > > to >> > > >> >> > > > >>>> iterate through the metadata for the dominant case >> where >> > > the >> > > >> >> > > partition >> > > >> >> > > > >>>> had >> > > >> >> > > > >>>> no transactions at all. >> > > >> >> > > > >>>> >> > > >> >> > > > >>>> -- >> > > >> >> > > > >>>> Divij Vaidya >> > > >> >> > > > >>>> >> > > >> >> > > > >>>> >> > > >> >> > > > >>>> >> > > >> >> > > > >>>> On Tue, Jun 25, 2024 at 11:42 AM Kamal >> Chandraprakash < >> > > >> >> > > > >>>> kamal.chandraprak...@gmail.com> wrote: >> > > >> >> > > > >>>> >> > > >> >> > > > >>>> > Bump. Please review this proposal. >> > > >> >> > > > >>>> > >> > > >> >> > > > >>>> > >> > > >> >> > > > >>>> > On Mon, Jun 17, 2024 at 6:55 PM Kamal >> Chandraprakash < >> > > >> >> > > > >>>> > kamal.chandraprak...@gmail.com> wrote: >> > > >> >> > > > >>>> > >> > > >> >> > > > >>>> > > Divij, >> > > >> >> > > > >>>> > > >> > > >> >> > > > >>>> > > Thanks for the review! Updated the KIP with 1, >> 2, 3, >> > > >> and 4 >> > > >> >> > > review >> > > >> >> > > > >>>> > > comments. >> > > >> >> > > > >>>> > > >> > > >> >> > > > >>>> > > > 4. Potential alternative - Instead of having an >> > > >> algorithm >> > > >> >> > > where >> > > >> >> > > > we >> > > >> >> > > > >>>> > > traverse >> > > >> >> > > > >>>> > > across segment metadata and looking for >> > isTxnIdxEmpty >> > > >> flag, >> > > >> >> > > should >> > > >> >> > > > >>>> we >> > > >> >> > > > >>>> > > directly introduce a nextSegmentWithTrxInx() >> > function? >> > > >> This >> > > >> >> > > would >> > > >> >> > > > >>>> allow >> > > >> >> > > > >>>> > > implementers to optimize the otherwise linear >> scan >> > > >> across >> > > >> >> > > metadata >> > > >> >> > > > >>>> for >> > > >> >> > > > >>>> > all >> > > >> >> > > > >>>> > > segments by using techniques such as skip list >> etc. >> > > >> >> > > > >>>> > > >> > > >> >> > > > >>>> > > This is a good point to optimize the scan. We >> need >> > to >> > > >> >> maintain >> > > >> >> > > the >> > > >> >> > > > >>>> > > skip-list >> > > >> >> > > > >>>> > > for each leader-epoch. With unclean leader >> election, >> > > >> some >> > > >> >> > > brokers >> > > >> >> > > > >>>> may not >> > > >> >> > > > >>>> > > have >> > > >> >> > > > >>>> > > the complete lineage. This will expand the scope >> of >> > > the >> > > >> >> work. >> > > >> >> > > > >>>> > > >> > > >> >> > > > >>>> > > In this version, we plan to optimize only for the >> > > below >> > > >> 2 >> > > >> >> > cases: >> > > >> >> > > > >>>> > > >> > > >> >> > > > >>>> > > 1. A partition does not have the transaction >> index >> > for >> > > >> any >> > > >> >> of >> > > >> >> > > the >> > > >> >> > > > >>>> > uploaded >> > > >> >> > > > >>>> > > segments. >> > > >> >> > > > >>>> > > The individual log segments `isTxnIdxEmpty` >> flag >> > > can >> > > >> be >> > > >> >> > > reduced >> > > >> >> > > > >>>> to a >> > > >> >> > > > >>>> > > single flag >> > > >> >> > > > >>>> > > in RLMM (using AND operator) that can serve >> the >> > > >> query - >> > > >> >> "Is >> > > >> >> > > all >> > > >> >> > > > >>>> the >> > > >> >> > > > >>>> > > transaction indexes empty for a partition?". >> > > >> >> > > > >>>> > > If yes, then we can directly scan the >> local-log >> > for >> > > >> >> aborted >> > > >> >> > > > >>>> > > transactions. >> > > >> >> > > > >>>> > > 2. A partition is produced using the >> transactional >> > > >> >> producer. >> > > >> >> > The >> > > >> >> > > > >>>> > > assumption made is that >> > > >> >> > > > >>>> > > the transaction will either commit/rollback >> > within >> > > >> 15 >> > > >> >> > > minutes >> > > >> >> > > > >>>> > > (default transaction.max.timeout.ms = 15 >> mins), >> > > >> >> possibly >> > > >> >> > we >> > > >> >> > > > >>>> may have >> > > >> >> > > > >>>> > > to search only >> > > >> >> > > > >>>> > > a few consecutive remote log segments to >> collect >> > > the >> > > >> >> > aborted >> > > >> >> > > > >>>> > > transactions. >> > > >> >> > > > >>>> > > 3. A partition is being produced with both normal >> > and >> > > >> >> > > > transactional >> > > >> >> > > > >>>> > > producers. In this case, >> > > >> >> > > > >>>> > > we will be doing linear traversal. >> Maintaining a >> > > >> >> skip-list >> > > >> >> > > > might >> > > >> >> > > > >>>> > > improve the performance but >> > > >> >> > > > >>>> > > we delegate the RLMM implementation to >> users. If >> > > >> >> > implemented >> > > >> >> > > > >>>> > > incorrectly, then it can lead >> > > >> >> > > > >>>> > > to delivery of the aborted transaction >> records >> > to >> > > >> the >> > > >> >> > > > consumer. >> > > >> >> > > > >>>> > > >> > > >> >> > > > >>>> > > I notice two drawbacks with the reduction method >> as >> > > >> >> proposed >> > > >> >> > in >> > > >> >> > > > the >> > > >> >> > > > >>>> KIP: >> > > >> >> > > > >>>> > > >> > > >> >> > > > >>>> > > 1. Even if one segment has a transaction index, >> then >> > > we >> > > >> >> have >> > > >> >> > to >> > > >> >> > > > >>>> iterate >> > > >> >> > > > >>>> > > over all the metadata events. >> > > >> >> > > > >>>> > > 2. Assume that there are 10 segments and >> segment-5 >> > > has a >> > > >> >> txn >> > > >> >> > > > index. >> > > >> >> > > > >>>> Once >> > > >> >> > > > >>>> > > the first 6 segments are deleted, >> > > >> >> > > > >>>> > > due to breach by time/size/start-offset, >> then we >> > > >> should >> > > >> >> > > return >> > > >> >> > > > >>>> `true` >> > > >> >> > > > >>>> > > for "Is all the transaction indexes empty for a >> > > >> partition?" >> > > >> >> > > > >>>> > > query but it will return `false` until the >> broker >> > > >> gets >> > > >> >> > > > restarted >> > > >> >> > > > >>>> and >> > > >> >> > > > >>>> > we >> > > >> >> > > > >>>> > > have to resort to iterate over all the metadata >> > > events. >> > > >> >> > > > >>>> > > >> > > >> >> > > > >>>> > > > 5. Potential alternative#2 - We know that we >> may >> > > want >> > > >> the >> > > >> >> > > > indexes >> > > >> >> > > > >>>> of >> > > >> >> > > > >>>> > > multiple higher segments. Instead of fetching >> them >> > > >> >> > sequentially, >> > > >> >> > > > we >> > > >> >> > > > >>>> could >> > > >> >> > > > >>>> > > implement a parallel fetch or a pre-fetch for the >> > > >> indexes. >> > > >> >> > This >> > > >> >> > > > >>>> would >> > > >> >> > > > >>>> > help >> > > >> >> > > > >>>> > > hide the latency of sequentially fetching the trx >> > > >> indexes. >> > > >> >> > > > >>>> > > >> > > >> >> > > > >>>> > > We can implement parallel-fetch/prefetch once the >> > > tiered >> > > >> >> > storage >> > > >> >> > > > is >> > > >> >> > > > >>>> GAed. >> > > >> >> > > > >>>> > > Since this feature will be useful >> > > >> >> > > > >>>> > > to prefetch the next remote log segment and it >> > expands >> > > >> the >> > > >> >> > scope >> > > >> >> > > > of >> > > >> >> > > > >>>> the >> > > >> >> > > > >>>> > > work. >> > > >> >> > > > >>>> > > >> > > >> >> > > > >>>> > > > 6. Should the proposed API take "segmentId" as >> a >> > > >> >> parameter >> > > >> >> > > > >>>> instead of >> > > >> >> > > > >>>> > > "topicIdPartition"? Suggesting because >> isTxnIdEmpty >> > is >> > > >> not >> > > >> >> a >> > > >> >> > > > >>>> property of >> > > >> >> > > > >>>> > a >> > > >> >> > > > >>>> > > partition, instead it's a property of a specific >> > > >> segment. >> > > >> >> > > > >>>> > > >> > > >> >> > > > >>>> > > We propose to use the `topicIdPartition` in >> > > >> >> > > > >>>> RemoteLogMetadataManager. >> > > >> >> > > > >>>> > > The implementation can fold/reduce the value of >> the >> > > >> >> individual >> > > >> >> > > log >> > > >> >> > > > >>>> > segment >> > > >> >> > > > >>>> > > `isTxnIdEmpty` flag. This is added to avoid >> scanning >> > > all >> > > >> >> the >> > > >> >> > > > >>>> metadata >> > > >> >> > > > >>>> > > events >> > > >> >> > > > >>>> > > when the partition does not have a transaction >> index >> > > in >> > > >> >> any of >> > > >> >> > > the >> > > >> >> > > > >>>> > > segments. >> > > >> >> > > > >>>> > > >> > > >> >> > > > >>>> > > On Mon, Jun 17, 2024 at 4:05 PM Divij Vaidya < >> > > >> >> > > > >>>> divijvaidy...@gmail.com> >> > > >> >> > > > >>>> > > wrote: >> > > >> >> > > > >>>> > > >> > > >> >> > > > >>>> > >> Hi Kamal >> > > >> >> > > > >>>> > >> >> > > >> >> > > > >>>> > >> Thanks for bringing this up. This is a problem >> > worth >> > > >> >> solving. >> > > >> >> > > We >> > > >> >> > > > >>>> have >> > > >> >> > > > >>>> > >> faced >> > > >> >> > > > >>>> > >> this in situations where some Kafka clients >> default >> > > to >> > > >> >> > > > >>>> read_committed >> > > >> >> > > > >>>> > mode >> > > >> >> > > > >>>> > >> and end up having high latencies for remote >> fetches >> > > >> due to >> > > >> >> > this >> > > >> >> > > > >>>> > traversal >> > > >> >> > > > >>>> > >> across all segments. >> > > >> >> > > > >>>> > >> >> > > >> >> > > > >>>> > >> First some nits to clarify the KIP: >> > > >> >> > > > >>>> > >> 1. The motivation should make it clear that >> > traversal >> > > >> of >> > > >> >> all >> > > >> >> > > > >>>> segments is >> > > >> >> > > > >>>> > >> only in the worst case. If I am not mistaken >> > (please >> > > >> >> correct >> > > >> >> > me >> > > >> >> > > > if >> > > >> >> > > > >>>> > wrong), >> > > >> >> > > > >>>> > >> the traversal stops when it has found a segment >> > > >> containing >> > > >> >> > LSO. >> > > >> >> > > > >>>> > >> 2. There is nothing like a non-txn topic. A >> > > transaction >> > > >> >> may >> > > >> >> > be >> > > >> >> > > > >>>> started >> > > >> >> > > > >>>> > on >> > > >> >> > > > >>>> > >> any topic. Perhaps, rephrase the statement in >> the >> > KIP >> > > >> so >> > > >> >> that >> > > >> >> > > it >> > > >> >> > > > is >> > > >> >> > > > >>>> > clear >> > > >> >> > > > >>>> > >> to the reader. >> > > >> >> > > > >>>> > >> 3. The hyperlink in the "the broker has to >> traverse >> > > all >> > > >> >> > the..." >> > > >> >> > > > >>>> seems >> > > >> >> > > > >>>> > >> incorrect. Did you want to point to >> > > >> >> > > > >>>> > >> >> > > >> >> > > > >>>> > >> >> > > >> >> > > > >>>> > >> > > >> >> > > > >>>> >> > > >> >> > > > >> > > >> >> > > >> > > >> >> > >> > > >> >> >> > > >> >> > > >> > >> https://github.com/apache/kafka/blob/21d60eabab8a14c8002611c65e092338bf584314/core/src/main/scala/kafka/log/LocalLog.scala#L444 >> > > >> >> > > > >>>> > >> ? >> > > >> >> > > > >>>> > >> 4. In the testing section, could we add a test >> > plan? >> > > >> For >> > > >> >> > > > example, I >> > > >> >> > > > >>>> > would >> > > >> >> > > > >>>> > >> list down adding a test which would verify the >> > number >> > > >> of >> > > >> >> > calls >> > > >> >> > > > >>>> made to >> > > >> >> > > > >>>> > >> RLMM. This test would have a higher number of >> calls >> > > >> >> earlier >> > > >> >> > vs. >> > > >> >> > > > >>>> after >> > > >> >> > > > >>>> > this >> > > >> >> > > > >>>> > >> KIP. >> > > >> >> > > > >>>> > >> >> > > >> >> > > > >>>> > >> Other thoughts: >> > > >> >> > > > >>>> > >> 4. Potential alternative - Instead of having an >> > > >> algorithm >> > > >> >> > where >> > > >> >> > > > we >> > > >> >> > > > >>>> > >> traverse >> > > >> >> > > > >>>> > >> across segment metadata and looking for >> > isTxnIdxEmpty >> > > >> >> flag, >> > > >> >> > > > should >> > > >> >> > > > >>>> we >> > > >> >> > > > >>>> > >> directly introduce a nextSegmentWithTrxInx() >> > > function? >> > > >> >> This >> > > >> >> > > would >> > > >> >> > > > >>>> allow >> > > >> >> > > > >>>> > >> implementers to optimize the otherwise linear >> scan >> > > >> across >> > > >> >> > > > metadata >> > > >> >> > > > >>>> for >> > > >> >> > > > >>>> > all >> > > >> >> > > > >>>> > >> segments by using techniques such as skip list >> etc. >> > > >> >> > > > >>>> > >> 5. Potential alternative#2 - We know that we may >> > want >> > > >> the >> > > >> >> > > indexes >> > > >> >> > > > >>>> of >> > > >> >> > > > >>>> > >> multiple higher segments. Instead of fetching >> them >> > > >> >> > > sequentially, >> > > >> >> > > > we >> > > >> >> > > > >>>> > could >> > > >> >> > > > >>>> > >> implement a parallel fetch or a pre-fetch for >> the >> > > >> indexes. >> > > >> >> > This >> > > >> >> > > > >>>> would >> > > >> >> > > > >>>> > help >> > > >> >> > > > >>>> > >> hide the latency of sequentially fetching the >> trx >> > > >> indexes. >> > > >> >> > > > >>>> > >> 6. Should the proposed API take "segmentId" as a >> > > >> parameter >> > > >> >> > > > instead >> > > >> >> > > > >>>> of >> > > >> >> > > > >>>> > >> "topicIdPartition"? Suggesting because >> isTxnIdEmpty >> > > is >> > > >> >> not a >> > > >> >> > > > >>>> property >> > > >> >> > > > >>>> > of a >> > > >> >> > > > >>>> > >> partition, instead it's a property of a specific >> > > >> segment. >> > > >> >> > > > >>>> > >> >> > > >> >> > > > >>>> > >> Looking forward to hearing your thoughts about >> the >> > > >> >> > > alternatives. >> > > >> >> > > > >>>> Let's >> > > >> >> > > > >>>> > get >> > > >> >> > > > >>>> > >> this fixed. >> > > >> >> > > > >>>> > >> >> > > >> >> > > > >>>> > >> -- >> > > >> >> > > > >>>> > >> Divij Vaidya >> > > >> >> > > > >>>> > >> >> > > >> >> > > > >>>> > >> >> > > >> >> > > > >>>> > >> >> > > >> >> > > > >>>> > >> On Mon, Jun 17, 2024 at 11:40 AM Kamal >> > > Chandraprakash < >> > > >> >> > > > >>>> > >> kamal.chandraprak...@gmail.com> wrote: >> > > >> >> > > > >>>> > >> >> > > >> >> > > > >>>> > >> > Hi all, >> > > >> >> > > > >>>> > >> > >> > > >> >> > > > >>>> > >> > I have opened a KIP-1058 >> > > >> >> > > > >>>> > >> > < >> > > >> >> > > > >>>> > >> > >> > > >> >> > > > >>>> > >> >> > > >> >> > > > >>>> > >> > > >> >> > > > >>>> >> > > >> >> > > > >> > > >> >> > > >> > > >> >> > >> > > >> >> >> > > >> >> > > >> > >> https://cwiki.apache.org/confluence/display/KAFKA/KIP-1058%3A+Txn+consumer+exerts+pressure+on+remote+storage+when+reading+non-txn+topic >> > > >> >> > > > >>>> > >> > > >> > > >> >> > > > >>>> > >> > to reduce the pressure on remote storage when >> > > >> >> transactional >> > > >> >> > > > >>>> consumers >> > > >> >> > > > >>>> > >> are >> > > >> >> > > > >>>> > >> > reading non-txn topics from remote storage. >> > > >> >> > > > >>>> > >> > >> > > >> >> > > > >>>> > >> > >> > > >> >> > > > >>>> > >> > >> > > >> >> > > > >>>> > >> >> > > >> >> > > > >>>> > >> > > >> >> > > > >>>> >> > > >> >> > > > >> > > >> >> > > >> > > >> >> > >> > > >> >> >> > > >> >> > > >> > >> https://cwiki.apache.org/confluence/display/KAFKA/KIP-1058%3A+Txn+consumer+exerts+pressure+on+remote+storage+when+reading+non-txn+topic >> > > >> >> > > > >>>> > >> > >> > > >> >> > > > >>>> > >> > Feedbacks and suggestions are welcome. >> > > >> >> > > > >>>> > >> > >> > > >> >> > > > >>>> > >> > Thanks, >> > > >> >> > > > >>>> > >> > Kamal >> > > >> >> > > > >>>> > >> > >> > > >> >> > > > >>>> > >> >> > > >> >> > > > >>>> > > >> > > >> >> > > > >>>> > >> > > >> >> > > > >>>> >> > > >> >> > > > >>> >> > > >> >> > > > >> > > >> >> > > >> > > >> >> > >> > > >> >> >> > > >> > >> > > >> >> > > > >> > > >> > >> >
