Fix Missing Re-Interrupts

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/core/KafkaAdmin.java

@@ -208,12 +208,15 @@ public final boolean initialize() {
 					addOrModifyTopicsIfNeeded(adminClient, newTopics);
 					return true;
 				}
-				catch (Exception e) {
+				catch (InterruptedException ex) {
+					Thread.currentThread().interrupt();
+				}
+				catch (Exception ex) {
 					if (!this.initializingContext || this.fatalIfBrokerNotAvailable) {
-						throw new IllegalStateException("Could not configure topics", e);
+						throw new IllegalStateException("Could not configure topics", ex);
 					}
 					else {
-						LOGGER.error(e, "Could not configure topics");
+						LOGGER.error(ex, "Could not configure topics");
 					}
 				}
 				finally {
@@ -267,6 +270,9 @@ public String clusterId() {
 			try (AdminClient client = createAdmin()) {
 				this.clusterId = client.describeCluster().clusterId().get(this.operationTimeout, TimeUnit.SECONDS);
 			}
+			catch (InterruptedException ex) {
+				Thread.currentThread().interrupt();
+			}
 			catch (Exception ex) {
 				LOGGER.error(ex, "Could not obtaine cluster info");
 			}

spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java

@@ -510,7 +510,11 @@ protected void checkTopics() {
 									entry.getValue().get(this.topicCheckTimeout, TimeUnit.SECONDS);
 									return false;
 								}
-								catch (@SuppressWarnings("unused") Exception e) {
+								catch (InterruptedException ex) {
+									Thread.currentThread().interrupt();
+									return true;
+								}
+								catch (@SuppressWarnings("unused") Exception ex) {
 									return true;
 								}
 							})