CAMEL-17489: camel-kafka - Unsubscribe before closing the consumer

[rgala]

No description provided.

[orpiske]

An exception here might trigger the exception handler ... but I guess it's OK.

[davsclaus]

I think safeUnsubscribe should be made safe and deal with already closed, eg kafka throws this

    private void acquireAndEnsureOpen() {
        acquire();
        if (this.closed) {
            release();
            throw new IllegalStateException("This consumer has already been closed.");
        }
    }

eg make safeUnsubscrive catch IllegalStateException and ignore that

[orpiske]

I think safeUnsubscribe should be made safe and deal with already closed, eg kafka throws this

    private void acquireAndEnsureOpen() {
        acquire();
        if (this.closed) {
            release();
            throw new IllegalStateException("This consumer has already been closed.");
        }
    }

eg make safeUnsubscrive catch IllegalStateException and ignore that

Yeah, I think it may be OK to ignore the IllegalStateException in this specific case. I think there are other issues that could happen during unsubscribe, so we need to be careful to ignore only that IMHO.

[orpiske]

I am merging this, and then I'll improve the safeUnsubscribe.