Fix Missing Re-Interrupts

Author: garyrussell

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

@@ -201,12 +201,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 {

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

@@ -488,7 +488,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;
 								}
 							})