Fix: don't set interrupt flag again after catching interrupt exception in Pulsar Client

[jerrypeng]

Motivation

Currently, if the a consumer gets interrupted will waiting on receive(), the interrupt exception just wrapped in a PulsarClientException and the thread is interrupted again. Setting the interrupt flag again for this thread is unnecessary as we already throw an exception and it can also unfavorable behavior. For example in the following code, if I have a thread...

Thread t = new Thread(new Runnable() {
            @Override
            public void run() {
                while(true) {
                    try {
                        consumer.receive();
                    } catch (Exception e) {
                        if (e instanceof PulsarClientException) {
                            break;
                        }
                    }
                }

                try {
                    consumer.close();
                    pulsarClient.close();
                } catch (Exception e) {
                    System.out.println("Closing error: " + e);
                }
            }
        });

The thread has a while loop that calls consumer receive(). If the thread gets interrupted, I would like to break from the while loop and close the consumer and client. However, closing the client and consumer cannot happen successfully because we set the interrupt flag for this thread after catching an interrupt exception causing any subsequent code e.g. close() to be interrupted as well which is not an ideal behavior.

This also causes non-ideal behavior in the pulsar-flink source as the above situation happens when a Flink job with a Pulsar source gets terminated.

Modifications

remove Thread.currentThread().interrupt();