This page is meant as a template for writing a KIP. To create a KIP choose Tools->Copy on this page and modify with your content and replace the heading with the next KIP number and a description of your issue. Replace anything in italics with your own description.

Status

Current state: Under Discussion

Discussion thread: here [Change the link from the KIP proposal email archive to your own email thread]

JIRA:

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

As we know, the callback of `Producer#send` is executed by the `ioThread`. This design has led to numerous application bugs caused by calling `Producer#flush` within the callback, resulting in deadlocks. Since `Producer#close` already includes logic to avoid blocking when invoked from a callback, we could implement similar logic for `Producer#flush`. This enhancement would help users avoid such pitfalls.

Public Interfaces

This KIP proposes two ways to implement this enhancement:

1. If the `Producer` detects that the `flush` method is invoked within the callback of the `close` method, log an error message.
2. If the `Producer` detects that the `flush` method is invoked within the callback of the `close` method, throw a `KafkaException`. This approach would require modifying org.apache.kafka.clients.producer.Producer#flush  to include KafkaException .

Proposed Changes

If we choose to follow point 2, we will need to make the following changes:

    /**
	 *                Omitting the unchange section ......
     * <p>
     * <b>Important:</b> This method must not be called from within the callback provided to
     * {@link #send(ProducerRecord, Callback)}.Invoking <code>flush()</code> in this context will result in a
     * {@link KafkaException} being thrown, as it will cause a deadlock.
     * </p>
     *
     * @throws InterruptException If the thread is interrupted while blocked
     * @throws KafkaException If the method is invoked inside a {@link #send(ProducerRecord, Callback)} callback
     */
    @Override
    public void flush() {
        if (Thread.currentThread() == this.ioThread) {
            log.error("KafkaProducer.flush() invocation inside a callback is not permitted because it may lead to deadlock.");
            throw new KafkaException("KafkaProducer.flush() invocation inside a callback is not permitted because it may lead to deadlock.");
        }
		
		// Omitting the unchange section ......

    }

Compatibility, Deprecation, and Migration Plan

N/A

Test Plan

Add a unit test to ensure that the deadlock protection throws a `KafkaException` with the corresponding error message to inform the user.

Rejected Alternatives

N/A