GH-1482: Make Topic Config Update Optional

Resolves #1482

Author: garyrussell

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

@@ -92,6 +92,8 @@ public class KafkaAdmin extends KafkaResourceFactory
 
 	private boolean initializingContext;
 
+	private boolean modifyTopicConfigs;
+
 	/**
 	 * Create an instance with an {@link AdminClient} based on the supplied
 	 * configuration.
@@ -140,6 +142,16 @@ public void setAutoCreate(boolean autoCreate) {
 		this.autoCreate = autoCreate;
 	}
 
+	/**
+	 * Set to true to compare the current topic configuration properties with those in the
+	 * {@link NewTopic} bean, and update if different.
+	 * @param modifyTopicConfigs true to check and update configs if necessary.
+	 * @since 2.8.7
+	 */
+	public void setModifyTopicConfigs(boolean modifyTopicConfigs) {
+		this.modifyTopicConfigs = modifyTopicConfigs;
+	}
+
 	@Override
 	public Map<String, Object> getConfigurationProperties() {
 		Map<String, Object> configs2 = new HashMap<>(this.configs);
@@ -254,16 +266,19 @@ private void addOrModifyTopicsIfNeeded(AdminClient adminClient, Collection<NewTo
 			if (topicsWithPartitionMismatches.size() > 0) {
 				createMissingPartitions(adminClient, topicsWithPartitionMismatches);
 			}
-			Map<ConfigResource, List<ConfigEntry>> mismatchingConfigs =
-					checkTopicsForConfigMismatches(adminClient, topics);
-			if (!mismatchingConfigs.isEmpty()) {
-				adjustConfigMismatches(adminClient, topics, mismatchingConfigs);
+			if (this.modifyTopicConfigs) {
+				Map<ConfigResource, List<ConfigEntry>> mismatchingConfigs =
+						checkTopicsForConfigMismatches(adminClient, topics);
+				if (!mismatchingConfigs.isEmpty()) {
+					adjustConfigMismatches(adminClient, topics, mismatchingConfigs);
+				}
 			}
 		}
 	}
 
 	private Map<ConfigResource, List<ConfigEntry>> checkTopicsForConfigMismatches(
 			AdminClient adminClient, Collection<NewTopic> topics) {
+
 		List<ConfigResource> configResources = topics.stream()
 				.map(topic -> new ConfigResource(Type.TOPIC, topic.name()))
 				.collect(Collectors.toList());
@@ -297,10 +312,10 @@ private Map<ConfigResource, List<ConfigEntry>> checkTopicsForConfigMismatches(
 		}
 		catch (InterruptedException ie) {
 			Thread.currentThread().interrupt();
-			throw new KafkaException("Interrupted while getting topic descriptions", ie);
+			throw new KafkaException("Interrupted while getting topic descriptions:" + topics, ie);
 		}
 		catch (ExecutionException | TimeoutException ex) {
-			throw new KafkaException("Failed to obtain topic descriptions", ex);
+			throw new KafkaException("Failed to obtain topic descriptions:" + topics, ex);
 		}
 	}
 

spring-kafka/src/test/java/org/springframework/kafka/core/KafkaAdminTests.java

@@ -106,6 +106,7 @@ public void testAddTopicsAndAddPartitions() throws Exception {
 		new DirectFieldAccessor(this.topic1).setPropertyValue("numPartitions", Optional.of(4));
 		new DirectFieldAccessor(this.topic2).setPropertyValue("numPartitions", Optional.of(3));
 		this.admin.initialize();
+		this.admin.setModifyTopicConfigs(true);
 
 		int n = 0;
 		await().until(() -> {
@@ -212,11 +213,12 @@ public void alreadyExists() throws Exception {
 		AtomicReference<Method> addTopics = new AtomicReference<>();
 		AtomicReference<Method> modifyTopics = new AtomicReference<>();
 		ReflectionUtils.doWithMethods(KafkaAdmin.class, m -> {
-			m.setAccessible(true);
 			if (m.getName().equals("addTopics")) {
+				m.setAccessible(true);
 				addTopics.set(m);
 			}
 			else if (m.getName().equals("createMissingPartitions")) {
+				m.setAccessible(true);
 				modifyTopics.set(m);
 			}
 		});