KAFKA-10188: Prevent SinkTask::preCommit from being called after SinkTask::stop

[C0urante]

Jira

The general lifecycle for a sink task is:

1. Instantiate the SinkTask object
2. Invoke SinkTask::initialize
3. Invoke SinkTask::start
4. While the task is still running:

• Poll Kafka for records
• Give those records to the task via SinkTask::put
• Periodically commit offsets, which involves calling SinkTask::preCommit and committing the resulting map of TopicPartition to offset to Kafka

5. Commit offsets a penultimate time (including the call to `SinkTask::preCommit)
6. Invoke SinkTask::stop
7. Close the consumer for the task
8. Commit offsets a final time (also including the call to SinkTask::preCommit)

This final offset commit happens indirectly: closing the consumer for a sink task causes the rebalance listener for that consumer to be triggered, and the rebalance listener the framework uses for its consumers performs an offset commit for the task when partitions are revoked.

This is a bit of a problem because the framework calls SinkTask::stop before closing the consumer for the task. It's possible and even likely that tasks will have de-allocated resources necessary for their preCommit method and will fail unexpectedly at this point.

Since the framework already ensures that offsets are committed after the last call to SinkTask::put, it should be fine to remove this extra offset commit. There is still a chance that some data may be dropped in the case that a task performs completely asynchronous writes to Kafka and has written data between the pre-stop call to SinkTask::preCommit and the post-stop one, but there will be no loss of delivery guarantees provided by the framework, and this change will adhere to the publicly-stated API for sink tasks.

A unit test is added that covers the internal WorkerSinkTask::close method and ensures that SinkTask::preCommit is not called during that method.

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[gharris1727]

LGTM, thanks @C0urante for the fix!

[rhauch]

Nice work, @C0urante. One question below.

[rhauch]

Shouldn't this be volatile?

Yes, it's true that WorkerSinkTask.close() is always and only called from within the WorkerTask.doRun() after the tasks determines it will stop. However, the onPartitionsRevoked(...) method is called from the consumer thread, and making the field volatile is the only way to ensure that the consumer thread reads a non-cached value.

[C0urante]

The Javadocs for the ConsumerRebalanceLister state that the callback "will only execute in the user thread as part of the poll(long) call"; I think we have a guarantee here that onPartitionsRevoked will be called on the same thread that sets taskStopped to false. A fun way to verify this is to view the exceptions that get thrown by this bug; the stack traces include these lines:

	at org.apache.kafka.connect.runtime.WorkerSinkTask$HandleRebalance.onPartitionsRevoked(WorkerSinkTask.java:695)
	at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.invokePartitionsRevoked(ConsumerCoordinator.java:312)
	at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onLeavePrepare(ConsumerCoordinator.java:744)
	at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.close(AbstractCoordinator.java:976)
	at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.close(ConsumerCoordinator.java:888)
	at org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2368)
	at org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2335)
	at org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2285)
	at org.apache.kafka.common.utils.Utils.closeQuietly(Utils.java:933)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.close(WorkerSinkTask.java:174)
	at org.apache.kafka.connect.runtime.WorkerTask.doClose(WorkerTask.java:164)
	at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:191)
	at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:235)

The only edge case I can think of might be with asynchronous offset commits, but fwict those don't trigger asynchronous rebalance listener callbacks (if they trigger rebalances or rebalance listener callbacks at all).

[rhauch]

Verified locally that this test fails when the additions to the onPartitionsRevoked(...) method above are removed locally. Nice work, @C0urante.