KAFKA-10340: Proactively close producer when cancelling source tasks

[C0urante]

Jira

When a source task produces records for a topic that doesn't exist on the Kafka cluster and automatic topic creation is disabled on the broker and not configured à la KIP-158 on the connector, the task hangs indefinitely until and unless the topic is created. Even if the task is scheduled for shutdown by the worker, it continues to hang; the SourceTask instance isn't stopped and the producer isn't closed.

One possible approach to handle this situation is to proactively close the producer for the task when it is abandoned after exceeding the graceful shutdown timeout period. This can increase the likelihood of duplicate records for tasks that are blocked on shutdown for other reasons (high throughput, for example), as offsets will not be committed for any outstanding batches. However, given that the Connect framework's overall delivery guarantees of Connect source connectors still remain intact with this approach (either at-least-once or at-most-once, but not exactly-once), the tradeoff seems acceptable in order to prevent resource leaks that, if stacked over a long enough period, will require worker restarts to deal with.

An existing unit test for the WorkerSourceTask class is expanded to ensure that the producer is closed when the task is abandoned, and a new integration test is added that guarantees that tasks are still shut down even when their producers are trying to write to topics that do not exist.

Committer Checklist (excluded from commit message)

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

[C0urante]

@gharris1727 @ncliang @SajanaW would any of you have time to take a look?

[C0urante]

Thanks for taking a look @gharris1727. I've tried to address your comments; let me know what you think.

[gharris1727]

wow, casting Runnable::run to Executor is pretty clever.

LGTM, thanks @C0urante!

[rhauch]

Thanks, @C0urante. Overall, this looks great. This PR has a merge conflict that needs to be addressed, and I do have a few other pretty minor suggestions below.

[rhauch]

The producer's close(...) method can throw an InterruptException if the method fails to join the IO thread. This can theoretically happen even if the timeout is 0 if the thread is interrupted (e.g., the executor is shutdown) before the join can wait. Although the likelihood of this is small, what do you think about catching InterruptException and ignoring the error?

Suggested change

	} catch (Throwable t) {
	} catch (InterruptException t) {
	// ignore, since this is likely due to the worker's executor being shut down
	} catch (Throwable t) {

Two things. First, the producer throws InterruptExeption, not InterruptedException. Second, even though the WorkerSourceTask::close() that calls this closeProducer(Duration) method doesn't directly use the executor, the Worker does use that same executor to stop this WorkerSourceTask, which ultimately does call WorkerSourceTask::close(). IOW, this closeProducer(Duration) method is always called from the executor, and the executor could be shutdown at any moment, thus the potential InterruptException.

[C0urante]

Right now there aren't any code paths that lead to the worker's executor being shut down. One might be added in the future, but it seems a little premature to try to catch it right now and may mislead readers of the code base. Plus, if an InterruptException does get generated somehow, it might be worth knowing about as it may indicate unhealthy (or at least unexpected) behavior from the worker, the task, an interceptor, etc.

[rhauch]

Right now there aren't any code paths that lead to the worker's executor being shut down.

Hmm, that seems to have been done a long time ago. I wonder if that was an oversight, or whether that was intentional since in Connect the Worker::stop() is called when the herder is stopped, which only happens (in Connect) when the shutdown hook is called -- at which point the JVM is terminating anyway. Luckily MM2 works the same way.

But in our test cases that use EmbeddedConnectCluster, those tests are not cleaning up all resources of the Worker (and thus Herder) -- we might have threads that still keep running. Seems like we should address that in a different issue. I'll log something.

[rhauch]

I've logged https://issues.apache.org/jira/browse/KAFKA-12380 for shutting down the worker's executor. Again, it's not an issue in runtime, but a potential issue in our tests.

[C0urante]

Thanks @rhauch. I've added all of your suggested changes except for the special handling for InterruptException (which I've responded to via comment) and fixed the merge conflict. Ready for another round when you have a chance.