Should <= be used ?

agreed

deliveryTimeoutMs should be mentioned

It seems better to say Producer.send() instead of send.

We are passing now everywhere else. Maybe we can just keep the argument name the same.

The actual argument is now. However, I like the formal argument name to be createTime because it's an immutable value while constructing a batch. now, is by definition, changing.

Should we validate the delivery.timeout.ms is greater than request.timeout.ms?

It is probably cleaner to have an explicit EXPIRED state.

I did some digging around. An expired batch's final state is FAILED. I don't feel great about adding yet another finalState. We already have ABORTED and FAILED. The ProducerBatch.done will get even more complicated.

Maybe it's not a big deal but just want to call out that this is a behavior change. Currently the producer will throw exception when transition from FAILED state to another state due to some reason other than expiration. If we change this logic, we may miss those cases which are not failed by expiration but still got state update twice. It may not be that important if we do not have programming bugs.

Personally I think it is better to clearly define the states of the batches even if additional complexity is necessary.

The comments should probably also cover the force close case for completeness.

The logic here probably needs more comments. We may have the following three cases that the state of a batch has been updated before the ProduceResponse returns:

1. A transaction abortion happens. The state of the batches would have been updated to ABORTED.
2. The producer is closed forcefully. The state of the batches would have been updated to ABORTED.
3. The batch is expired when it is in-flight. The state of the batch would have been updated to EXPIRED.

In the other cases, we should throw IllegalStateException.

Please review the updated method documentation.

The batches still needs to be expired in order if max.in.flight.requests.per.connection is set to 1. So we probably still want to check if the partition is muted or not. That said, if we guarantee that when RecordAccumulator.expiredBatches() returns non-empty list, all the earlier batches have already been expired, we can remove the muted check here.

BTW, I did not see the logic of expiring an in-flight batch in the current patch. Am I missing something?

It's there now

@becketqin Added the muted check back

isFull is no longer used.

Still not used.

This variable can be dropped.

in not -> is not

The check 'if (deliveryTimeoutMs <= (now - this.createdMs))' inside maybeExpire() would be true.
Looks like another method can be created inside ProducerBatch which expires the batch.

maybeExpire has a side-effect of setting errorMessage internally. Hence calling it again in if.

Understand.
That part can be refactored - goal is to reduce unnecessary comparison.

@apurvam Those test don't even compile or run on my machine. What's up with those tests?

They can't construct a kafka producer with the changes made in this PR.

Assuming nFlightBatches is a TreeSet suggested above, this code can be simplified to:

        while (!inFlightBatches.isEmpty() &&
               inFlightBatches.first().maybeExpire(deliveryTimeoutMs, now)) {
            expiredBatches.add(inFlightBatches.pollFirst());
        }

Is this comment accurate? The new state is not necessarily SUCCEEDED.

Maybe it's not a big deal but just want to call out that this is a behavior change. Currently the producer will throw exception when transition from FAILED state to another state due to some reason other than expiration. If we change this logic, we may miss those cases which are not failed by expiration but still got state update twice. It may not be that important if we do not have programming bugs.

Personally I think it is better to clearly define the states of the batches even if additional complexity is necessary.

The comments should probably also cover the force close case for completeness.

Some typos in this comments. "Expire the batch if no outcome is known within delivery.timeout.ms"

Does this have to be a per partition Map? Intuitively we just need a TreeSet<ProducerBatch> with a comparator?

Apparently the my understanding of TreeSet is not accurate. It uses the comparator to decide whether the entries are the same or not. We can use a TreeMap<Long, Set> then. We may also want to bucket the timestamp a little bit, say one second to avoid huge amount of Sets created for each ms in the TreeMap.

I was thinking about this too. Using millisecond as unit for Map key is not prudent.

After the switch to second as unit, we may need to check the two adjacent buckets keyed by ts-1 (sec) and ts+1 (sec).

As we discussed, TreeSet does not cut it. The naming is consistent. A TreeSet is a set. It's just that equality criterion is different.

Assuming nFlightBatches is a TreeSet suggested above, this code can be simplified to:

        while (!inFlightBatches.isEmpty() &&
               inFlightBatches.first().maybeExpire(deliveryTimeoutMs, now)) {
            expiredBatches.add(inFlightBatches.pollFirst());
        }

This logic would become inFlightRequests.remove(batch) when a TreeSet is used for this.

This would be just inFlightBatches.add(batch)

We usually just use earliestDeliveryTimeout in Kafka.

It seems we don't need the deliveryTimeoutMs in the sender. It is only used as an argument passed to the accumulator. But the accumulator already has the config.

It seems an existing issue. When we expire the batches here. The memory of those batches will be deallocated. It seems that we will deallocate the same batch again when the ProduceResponse returns?

This test has nothing to do with linger.ms anymore...

We should change the test name to something like testBatchExpiration. and the test below to testBatchExpirationAfterReenqueue.

Similar to above we should rename this.

typo: timeout

The typo is still there.

typo: timeout

Should we still expire the batches when they are expired instead of expiring all the bucket? Having a second granularity bucket does not prevent us from doing that, right?

Still not used.

We don't need a PriorityQueue for this because the batches in the RecordAccumulator is already in order. So we just need to keep the draining order.

If we always insert the batch to the inFlightBatches queue and there is no bug, the batch to be removed should always be the first batch. Can we assert on that?

The original reason we have this optimization is because we used to have a big sorted data structure. So avoiding inserting elements to it makes sense. Given that now the batch order in the RecordAccumulator is already guaranteed. It seems we can just put all the drained batches to the inFlightBatches queue, which is simpler.

The while loop may break if the request size has reached. So there is no guarantee that it will iterate over all the partitions. One alternative is to find the nextBatchExpiryTimeMs in the expireBatches.

It seems intuitively this should be the earliest batch in the entire record accumulator?

It seems we may release the memory for the expired batches before the response is returned. This means the underneath ByteBuffer is still referred by the ProducerBatch instance in the inFlightRequests. I am not sure if this would cause any problem, but it seems a little dangerous.