Just to clarify on the implementation details: today inside
Sender#runOnce(), we have the following:
```
if (maybeSendAndPollTransactionalRequest()) {
return;
}
```
Which basically means that as long as we still have any in-flight txn
request; so we are not really blocking on the first sent request but on all
of the types. For add-partitions requests, that makes sense since once
unblocked, all requests would unblock at the same time anyways; but we do
not necessarily need to block on AddOffsetCommitsToTxnRequest and
TxnOffsetCommitRequest.
So we can relax the above restriction as 1) the txn coordinator is known,
2) the producer has a valid PID. 3) all partitions are registered in the
txn.
Guozhang
On Wed, May 20, 2020 at 3:13 PM Boyang Chen <reluctanthero...@gmail.com>
wrote:
> Thanks Guozhang for the new proposal. I agree that we could deliver
> https://issues.apache.org/jira/browse/KAFKA-9878
> <https://issues.apache.org/jira/browse/KAFKA-9878?src=confmacro> first and
> measure the following metrics:
>
> 1. The total volume of AddPartitionToTxn requests
> 2. The time used in propagating the transaction state updates during
> transaction
> 3. The time used in transaction marker propagation
>
> If those metrics suggest that we are doing a pretty good job already, then
> the improvement of delivering the entire KIP-609 is minimal. In the
> meantime, I updated 9878 with more details. Additionally, I realized that
> we should change the logic for AddPartitionToTxn call so that we could
> maintain a future queue and wait for all the delta change completions,
> instead of blocking on the first sent out one. Does that make sense?
>
> On Wed, May 20, 2020 at 2:28 PM Guozhang Wang <wangg...@gmail.com> wrote:
>
> > Hello Matthias,
> >
> > I have a quick question regarding the motivation of the long-blocking and
> > batch-add-partitions behavior: do we think the latency primarily comes
> from
> > the network round-trips, or from the coordinator-side transaction-log
> > appends? If we believe it is coming from the latter, then perhaps we can
> > first consider optimizing that without making any public changes,
> > specifically:
> >
> > 1) We block on the add-partitions in a purgatory, as proposed in your
> KIP.
> >
> > 2) When try-completing the parked add-partitions requests in a purgatory,
> > we consolidate them as a single txn state transition with a single append
> > to transaction log.
> >
> > 3) *Optionally* on the client side, we can further optimize the behavior:
> > instead of block on sending any batches as long as there are any txn
> > requests in flight, we would just query which partitions has successfully
> > "registered" as part of the txn from add-partitions response and then
> send
> > records for those partitions. By doing this we can reduce the end-to-end
> > blocking time.
> >
> > None of the above changes actually requires any public API or protocol
> > changes. In addition, it would not make things worse even in edge cases
> > whereas with the proposed API change, if the producer pre-registered a
> > bunch of partitions but then timed out, the coordinator need to write
> abort
> > markers to those pre-registered partitions unnecessarily. We can measure
> > the avg. number of txn log appends per transaction on the broker side and
> > see if it can be reduced to close to 1 already.
> >
> >
> > Guozhang
> >
> >
> > On Tue, May 19, 2020 at 10:33 AM Boyang Chen <reluctanthero...@gmail.com
> >
> > wrote:
> >
> > > Hey John,
> > >
> > > thanks for the insights! Replied inline.
> > >
> > > On Tue, May 19, 2020 at 7:48 AM John Roesler <vvcep...@apache.org>
> > wrote:
> > >
> > > > Thanks for the KIP, Boyang!
> > > >
> > > > This looks good and reasonable to me overall.
> > > >
> > > > J1: One clarification: you mention that the blocking behavior depends
> > on
> > > > a new broker version, which sounds good to me, but I didn't see why
> > > > we would need to throw any UnsupportedVersionExceptions. It sounds
> > > > a little like you just want to implement a kind of long polling on
> the
> > > > AddPartitionToTxn API, such that the broker would optimistically
> block
> > > > for a while when there is a pending prior transaction.
> > > >
> > > > Can this just be a behavior change on the broker side, such that both
> > > > old and new clients would be asked to retry when the broker is older,
> > > > and both old and new clients would instead see the API call block for
> > > > longer (but be successful more often) when the broker is newer?
> > > >
> > > > Related: is it still possible to get back the "please retry" error
> from
> > > the
> > > > broker, or is it guaranteed to block until the call completes?
> > > >
> > > > This is a good observation. I agree the blocking behavior could
> benefit
> > > all the producer
> > > versions older than 0.11, which could be retried. Added to the KIP.
> > >
> > >
> > > > J2: Please forgive my ignorance, but is there any ill effect if a
> > > producer
> > > > adds a partition to a transaction and then commits without having
> added
> > > > any data to the transaction?
> > > >
> > > > I can see this happening, e.g., if I know that my application
> generally
> > > > sends to 5 TopicPartitions, I would use the new beginTransaction call
> > > > and just always give it the same list of partitions, and _then_ do
> the
> > > > processing, which may or may not send data to all five potential
> > > > partitions.
> > > >
> > >
> > > Yes, that's possible, which is the reason why we discussed bumping the
> > > EndTxn request
> > > to only include the partitions actually being written to, so that the
> > > transaction coordinator will only send markers
> > > to the actually-written partitions. The worst case for mis-used
> > > pre-registration API
> > > is to write out more transaction markers than necessary. For once, I do
> > see
> > > the benefit of doing that,
> > > which is a life-saver for a "lazy user" who doesn't want to infer the
> > > output partitions it would write to, but always
> > > registers the full set of output partitions. With this observation in
> > mind,
> > > bumping EndTxn makes sense.
> > >
> > > >
> > > > Thanks again!
> > > > -John
> > > >
> > > > On Mon, May 18, 2020, at 10:25, Boyang Chen wrote:
> > > > > Oh, I see your point! Will add that context to the KIP.
> > > > >
> > > > > Boyang
> > > > >
> > > > > On Sun, May 17, 2020 at 11:39 AM Guozhang Wang <wangg...@gmail.com
> >
> > > > wrote:
> > > > >
> > > > > > My point here is only for the first AddPartitionToTxn request of
> > the
> > > > > > transaction, since only that request would potentially be blocked
> > on
> > > > the
> > > > > > previous txn to complete. By deferring it we reduce the blocking
> > > time.
> > > > > >
> > > > > > I think StreamsConfigs override the linger.ms to 100ms not 10ms,
> > so
> > > > in the
> > > > > > best case we can defer the first AddPartitionToTxn of the
> > transaction
> > > > by
> > > > > > 100ms.
> > > > > >
> > > > > > Guozhang
> > > > > >
> > > > > >
> > > > > > On Sat, May 16, 2020 at 12:20 PM Boyang Chen <
> > > > reluctanthero...@gmail.com>
> > > > > > wrote:
> > > > > >
> > > > > > > Thanks Guozhang for the context. The producer batch is either
> > > > bounded by
> > > > > > > the size or the linger time. For the default 10ms linger and
> > 100ms
> > > > > > > transaction commit time, the producer will be capped by
> > > > > > > AddPartitionToTxn 10 times in the worst case. I think the
> > > improvement
> > > > > > here
> > > > > > > aims for the worst case scenario for users who didn't realize
> how
> > > the
> > > > > > > internal works, and uses the API calls in a very inefficient
> way
> > as
> > > > the
> > > > > > > scenario where record processing and send() happen
> concurrently.
> > > > > > >
> > > > > > > Boyang
> > > > > > >
> > > > > > > On Sat, May 16, 2020 at 10:19 AM Guozhang Wang <
> > wangg...@gmail.com
> > > >
> > > > > > wrote:
> > > > > > >
> > > > > > > > Hello Boyang,
> > > > > > > >
> > > > > > > > Thanks for the proposed KIP, overall it makes sense to me.
> > > > > > > >
> > > > > > > > One non-public API related point that I'd like to make
> though,
> > is
> > > > that
> > > > > > in
> > > > > > > > KafkaProducer.send call we can potentially defer sending
> > > > > > > > AddPartitionsToTxn request until the sender is about to send
> > the
> > > > first
> > > > > > > > batch -- this is what I observed from some soak testing
> > > > investigation
> > > > > > > such
> > > > > > > > that the batching effects actually allows the first record to
> > be
> > > > sent
> > > > > > > much
> > > > > > > > later than the send() call and that can be leveraged to
> further
> > > > reduce
> > > > > > > the
> > > > > > > > time that we would be blocked on the AddPartitionsToTxn
> > request.
> > > > > > > >
> > > > > > > >
> > > > > > > > Guozhang
> > > > > > > >
> > > > > > > >
> > > > > > > > On Thu, May 14, 2020 at 10:26 PM Boyang Chen <
> > > > > > reluctanthero...@gmail.com
> > > > > > > >
> > > > > > > > wrote:
> > > > > > > >
> > > > > > > > > Hey all,
> > > > > > > > >
> > > > > > > > > I would like to start the discussion for KIP-609:
> > > > > > > > >
> > > > > > > > >
> > > > > > > >
> > > > > > >
> > > > > >
> > > >
> > >
> >
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-609%3A+Use+Pre-registration+and+Blocking+Calls+for+Better+Transaction+Efficiency
> > > > > > > > >
> > > > > > > > > This KIP aims to improve the current EOS semantic which
> makes
> > > the
> > > > > > > > > processing more efficient and consolidated.
> > > > > > > > >
> > > > > > > > > Thanks!
> > > > > > > > >
> > > > > > > >
> > > > > > > >
> > > > > > > > --
> > > > > > > > -- Guozhang
> > > > > > > >
> > > > > > >
> > > > > >
> > > > > >
> > > > > > --
> > > > > > -- Guozhang
> > > > > >
> > > > >
> > > >
> > >
> >
> >
> > --
> > -- Guozhang
> >
>
--
-- Guozhang
