This check is a little weird to me. It seems arbitrary to check only lingerMs in the case of overflow. Wouldn't it make more sense to use Integer.MAX_VALUE?

updated the sanity checking logic. changed lingerMs and deliveryTimeoutMs to integer type. with that, we convert them to long for addition and do not need to worry about the overflow.

I'm wondering if we should set this value automatically if the value is not overridden. It would be annoying to get this config error after an upgrade if I have overridden the request timeout. Maybe we could use the max of the default delivery timeout and the sum of requestTimeoutMs and lingerMs?

my concern is that setting the value automatically may hide some issues from the user, and it also requires more explanation. It might be better to keep it simple. The issue that the users can run into an upgrade is that they override requestTimeoutMs with a large value, and that violates the required variant deliveryTimeout > requestTimeoutMs + lingerMs. In that case, the user will be notify the error during producer initialization, and it shall not take much effort to fix it.

Yeah, I appreciate the concern. However, there is already some precedent for redefining defaults based on provided configurations. For example, when idempotence is enabled, we override retries if it has not been explicitly provided by the user. We log an info message so that the user knows the default has been adjusted.

In general, I think we should try to avoid compatibility issues even in configuration as long as it is safe to do so. By using a larger request timeout, the user has already declared willingness to await for the request timeout to receive a produce acknowledgement, so redefining the default delivery timeout seems reasonable and saves an annoying config update.

A reason for the error is that a longer request timeout is not unlikely to have been done as a workaround for the issue that delivery timeout is fixing. So, it might provide the user with an opportunity to fix that. A warning might be a less heavy handed way to achieve this though.

updated the change to use requestTimeoutMs + lingerMs if it is larger than deliveryTimeoutMs setting, and log a warning message.

The latter half here may not be very clear to users since it's a bit low level. Is there actually a lower bound for the timeout or can it be arbitrarily small depending on how long the batch remains open?

updated the doc string on the lower bound explanation.

Can you mention this default change in the upgrade notes?

updated the upgrade doc on this.

Maybe helpful to mention the baseOffset in this message?

added baseOffset info in the log message

nit: this is unused

removed

Can we explain in this comment why we use this order? Intuitively, I would expect that we'd expire the oldest stuff first so that the callbacks are invoked in the order of sending.

this is a comment that was in #3849. updated the comment.

Hmm.. Shouldn't we be expiring the batch in the else case? Otherwise it seems like we may lose track of the batch.

good catch. expire the batch in else branch.

Unless I'm missing something, we only remove from the inFlightBatches collection when we reenqueue and when we encounter a delivery timeout. Don't we need to remove on completion or failure as well?

Good catch. This would have to be called from Sender.failBatch and Sender.completeBatch as well, similar to how the transactionManager.removeInflightBatch is called in those methods.

good catch! updated the code to call maybeRemoveFromInflightBatches from Sender.failBatch and Sender.completeBatch.

nit: unintentional?

restored the space

In what scenario will batch.createdMs + deliveryTimeoutMs be negative? batch.createdMs is the wall clock time when the batch is created, and should always be positive. The config def for deliveryTimeoutMs enforces the value is atleast(0). So then how could the sum of the two be negative? Are you accounting for wrap around?

yes, this is to guard us against potential overflow due to setting a large value for deliveryTimeoutMs.

Ok. It would be good mention this in a comment. It is very non obvious.

update the code to use if ... statement and add the comment

I am assuming this method and drainBatchesForOneNode is just a refactor? Was any logic changed here? It is hard to tell from the diff and the logic is super intricate so subtle changes are easy to miss. I assume nothing should have to change in this portion of the code since it is totally orthogonal to expiring batches.

yes, this is a refactoring. otherwise, the method RecordAccumulator.drain fails in style checking due to too many possible paths in one method. comparing with the previous code, the change is to add the following lines on updating inflightBatches

               // put this batch in the infligh list
                List<ProducerBatch> inflightBatchList = inFlightBatches.get(batch.topicPartition);
                if (inflightBatchList == null) {
                    inflightBatchList = new LinkedList<>();
                    inFlightBatches.put(tp, inflightBatchList);
                }
                inflightBatchList.add(batch);

here and elsewhere, the code style for kafka requires that there be braces even around single line if statements like this.

restored the curly braces.

The preceding comment needs to be updated to account for this new logic.

good point. updated the comment

This seems not to be used.

good catch. removed the unused code.

Good catch. This would have to be called from Sender.failBatch and Sender.completeBatch as well, similar to how the transactionManager.removeInflightBatch is called in those methods.

nit: move this to a helper named markBatchInflight or something similar.

updated the code to capture this in markBatchInflight method

We should log a warning if we have overflowed and are hence not updating the next batch expiry time.

good point. added the logging here.

We should add some sort of accumulator.hasInflightBatches method, and then check that it returns false here. This would check that expired batches are not reenqueued, which is logic added in this patch.

added the method public List<ProducerBatch> inFlightBatches(TopicPartition tp) in RecordAccumulator, and updated the test with two assertions:

   line 1912: assertEquals("Expect one in-flight batch in accumulator", 1, accumulator.inFlightBatches(tp0).size());
    .....
   line 1920: assertEquals("Expect zero in-flight batch in accumulator", 0, accumulator.inFlightBatches(tp0).size());

How is this different from the test testExpiryOfFirstBatchShouldNotCauseUnresolvedSequencesIfFutureBatchesSucceed?

testExpiryOfFirstBatchShouldNotCauseUnresolvedSequencesIfFutureBatchesSucceed initialize sender with guaranteeMessageOrder = false, while testWhenFirstBatchExpireNoSendSecondBatchIfGuaranteeOrder initialize sender with guaranteeMessageOrder = true. The inflightBatches size is different when we set the parameter to true/false.

THis should be dropped, or be log.debug.

good catch. removed this debugging line.

If we had accumulator.hasInflightRequests we could assert false here.

How is this testing that we don't double deallocate. Ideally, we would expire the inflight batch, then get a response, and then check that the batch is not deallocated twice. In this case, it would de allocate only once, unless I am missing something.

updated the test a bit to add another sender.run(time.milliseconds()); call after the batch expiry. Not sure if that fits expire the inflight batch, then get a response.

my expectation was that if there is double deallocation, we would get an IllegalStateException exception from MatchingBufferPool.deallocate.

To clarify, I think we should only override the default value. If the user has explicitly provided an inconsistent value, then we should throw an exception. We can check this using config.originals().

updated

nit: createdMs for consistency?

By the way, we have a function below createdTimeMs which is currently unused and seems incorrect anyway. Can you remove it?

updated to createdMs, and removed the unused method.

nit: "Ignores" -> "Ignored"?

updated

It seems tryFinalState can only be SUCCEEDED or FAILED, so one of these transitions is not possible anyway.

ProducerBatcn.finalState can also be updated to FinalState.ABORTED through Sender.run() --> RecordAccumulator.abortIncompleteBatches() or abortUndrainedBatches() --> ... -> ProducerBatch.abort() .

My point is that tryFinalState can only be SUCCEEDED or FAILED.

final FinalState tryFinalState = (exception == null) ? FinalState.SUCCEEDED : FinalState.FAILED;

So the only possibilities are FAILED -> FAILED and ABORTED -> FAILED.

i see. misunderstood your comment earlier. updated the change

Are the order of the first two arguments backwards? I think tryFinalState is the state we're trying to transition to.

good catch! fixed the arguments order.