diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java
index 45e4f92d6..c06cef0ea 100644
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java
+++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java
@@ -64,6 +64,7 @@ import org.springframework.util.StringUtils;
  * @author Nico Heller
  * @author Norbert Gyurian
  * @author Boini Srinivas
+ * @author Felix Schultze
  */
 public class KafkaBinderConfigurationProperties {
 
@@ -211,7 +212,7 @@ public class KafkaBinderConfigurationProperties {
 		final String storeLocation = this.configuration.get(storeProperty);
 
 		// If the path is not defined, or it is a local file path do not move the file
-		if (storeLocation != null && !checkIfFileExists(storeLocation)) {
+		if (StringUtils.hasText(storeLocation) && !checkIfFileExists(storeLocation)) {
 			final String fileSystemLocation = moveCertToFileSystem(storeLocation, this.certificateStoreDirectory);
 			// Overriding the value with absolute filesystem path.
 			this.configuration.put(storeProperty, fileSystemLocation);
diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/test/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationPropertiesTest.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/test/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationPropertiesTest.java
index a7d405c28..3b77d5d96 100644
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/test/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationPropertiesTest.java
+++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/test/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationPropertiesTest.java
@@ -253,6 +253,22 @@ public class KafkaBinderConfigurationPropertiesTest {
 
 	}
 
+	@Test
+	public void testEmptyLocationsAreIgnored() {
+		KafkaProperties kafkaProperties = new KafkaProperties();
+		KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties =
+				new KafkaBinderConfigurationProperties(kafkaProperties);
+		final Map<String, String> configuration = kafkaBinderConfigurationProperties.getConfiguration();
+		configuration.put("schema.registry.ssl.truststore.location", "");
+		configuration.put("schema.registry.ssl.keystore.location", "");
+		kafkaBinderConfigurationProperties.setCertificateStoreDirectory("target");
+
+		kafkaBinderConfigurationProperties.getKafkaConnectionString();
+
+		assertThat(configuration.get("schema.registry.ssl.truststore.location")).isEmpty();
+		assertThat(configuration.get("schema.registry.ssl.keystore.location")).isEmpty();
+	}
+
 	private void createContextWithCertFileHandler(HttpServer server, String path) {
 		server.createContext("/" + path, exchange -> {
 			ClassPathResource ts = new ClassPathResource(path);