Quick question: any thoughts on a test we could write for this?

So, the thing to keep in mind here is that when the worker shuts itself down, its stop function will be called shortly thereafter. Since the stop function closes the consumer, there is no actual resource leak here.

One thing I notice that could be fixed is that ThreadSafeConsumer#close does not guard against unsubscribe throwing an exception. It would be good to add this in. Also, the locking is slightly wrong since ThreadSafeConsumer#closed is checked without holding the lock.

        void close() {
            if (closed)
                return;
            this.consumerLock.lock();
            try {
                consumer.unsubscribe();
                Utils.closeQuietly(consumer, "consumer");
                closed = true;
            } finally {
                this.consumerLock.unlock();
            }
        }

This PR is being marked as stale since it has not had any activity in 90 days. If you
would like to keep this PR alive, please leave a comment asking for a review. If the PR has
merge conflicts, update it with the latest from the base branch.

If you are having difficulty finding a reviewer, please reach out on the [mailing list](https://kafka.apache.org/contact).

If this PR is no longer valid or desired, please feel free to close it. If no activity occurs in the next 30 days, it will be automatically closed.

This PR has been closed since it has not had any activity in 120 days. If you feel like this
was a mistake, or you would like to continue working on it, please feel free to re-open the
PR and ask for a review.