diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java
index 4b7744f4b..e6fad5acc 100644
--- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java
@@ -44,6 +44,7 @@ import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.ApplicationEventPublisherAware;
 import org.springframework.context.Lifecycle;
 import org.springframework.integration.channel.AbstractMessageChannel;
+import org.springframework.integration.channel.AbstractSubscribableChannel;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.MessageChannelReactiveUtils;
 import org.springframework.integration.channel.PublishSubscribeChannel;
@@ -607,22 +608,46 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 			handler = getDefaultErrorMessageHandler((LastSubscriberAwareChannel) errorChannel, defaultErrorChannel != null);
 		}
 		String errorMessageHandlerName = getErrorMessageHandlerName(destination, group, consumerProperties);
+
 		if (handler != null) {
-			this.registerComponentWithBeanFactory(errorMessageHandlerName, handler);
-			beanFactory.initializeBean(handler, errorMessageHandlerName);
-			errorChannel.subscribe(handler);
+			if (this.isSubscribable(errorChannel)) {
+				this.registerComponentWithBeanFactory(errorMessageHandlerName, handler);
+				beanFactory.initializeBean(handler, errorMessageHandlerName);
+				errorChannel.subscribe(handler);
+			}
+			else {
+				logger.warn("The provided errorChannel '" + errorChannelName + "' is an instance of DirectChannel, "
+						+ "so no more subscribers could be added which may affect DLQ processing. Resolution: Configure your own errorChannel as "
+						+ "an instance of PublishSubscribeChannel");
+			}
 		}
+
 		if (defaultErrorChannel != null) {
-			BridgeHandler errorBridge = new BridgeHandler();
-			errorBridge.setOutputChannel(defaultErrorChannel);
-			errorChannel.subscribe(errorBridge);
-			String errorBridgeHandlerName = getErrorBridgeName(destination, group, consumerProperties);
-			this.registerComponentWithBeanFactory(errorBridgeHandlerName, errorBridge);
-			beanFactory.initializeBean(errorBridge, errorBridgeHandlerName);
+			if (this.isSubscribable(errorChannel)) {
+				BridgeHandler errorBridge = new BridgeHandler();
+				errorBridge.setOutputChannel(defaultErrorChannel);
+				errorChannel.subscribe(errorBridge);
+				String errorBridgeHandlerName = getErrorBridgeName(destination, group, consumerProperties);
+				this.registerComponentWithBeanFactory(errorBridgeHandlerName, errorBridge);
+				beanFactory.initializeBean(errorBridge, errorBridgeHandlerName);
+			}
+			else {
+				logger.warn("The provided errorChannel '" + errorChannelName + "' is an instance of DirectChannel, "
+						+ "so no more subscribers could be added and no error messages will be sent to global error channel. Resolution: Configure your own errorChannel as "
+						+ "an instance of PublishSubscribeChannel");
+			}
 		}
 		return new ErrorInfrastructure(errorChannel, recoverer, handler);
 	}
 
+	private boolean isSubscribable(SubscribableChannel errorChannel) {
+		if (errorChannel instanceof PublishSubscribeChannel) {
+			return true;
+		}
+		return errorChannel instanceof AbstractSubscribableChannel
+				? ((AbstractSubscribableChannel)errorChannel).getSubscriberCount() == 0 : true;
+	}
+
 	private void destroyErrorInfrastructure(ProducerDestination destination) {
 		String errorChannelName = errorsBaseName(destination);
 		String errorBridgeHandlerName = getErrorBridgeName(destination);
@@ -785,7 +810,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 
 	private SubscribableChannel postProcessOutboundChannelForFunction(MessageChannel outputChannel) {
 		if (this.integrationFlowFunctionSupport != null) {
-			Publisher publisher = MessageChannelReactiveUtils.toPublisher(outputChannel);
+			Publisher<?> publisher = MessageChannelReactiveUtils.toPublisher(outputChannel);
 			// If the app has an explicit Supplier bean defined, make that as the publisher
 			if (this.integrationFlowFunctionSupport.containsFunction(Supplier.class)) {
 				IntegrationFlowBuilder integrationFlowBuilder = IntegrationFlows.from(outputChannel).bridge();
diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ErrorBindingTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ErrorBindingTests.java
index bd872f87e..1f46a1781 100644
--- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ErrorBindingTests.java
+++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ErrorBindingTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -17,21 +17,35 @@
 package org.springframework.cloud.stream.binder;
 
 import org.junit.Test;
+
 import org.mockito.Mockito;
 
 import org.springframework.boot.SpringApplication;
+import org.springframework.boot.WebApplicationType;
 import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.builder.SpringApplicationBuilder;
 import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.annotation.StreamListener;
+import org.springframework.cloud.stream.binder.test.InputDestination;
+import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
 import org.springframework.cloud.stream.messaging.Processor;
+import org.springframework.cloud.stream.messaging.Sink;
+import org.springframework.context.ApplicationContext;
 import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.integration.annotation.ServiceActivator;
+import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.support.GenericMessage;
 
+
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.ArgumentMatchers.isNull;
 
 /**
  * @author Marius Bogoevici
+ * @author Oleg Zhurakousky
  */
 public class ErrorBindingTests {
 
@@ -39,7 +53,7 @@ public class ErrorBindingTests {
 	@Test
 	public void testErrorChannelNotBoundByDefault() {
 		ConfigurableApplicationContext applicationContext = SpringApplication.run(TestProcessor.class,
-				"--server.port=0", "--spring.cloud.stream.default-binder=mock");
+				"--server.port=0", "--spring.cloud.stream.default-binder=mock", "--spring.jmx.enabled=false");
 		BinderFactory binderFactory = applicationContext.getBean(BinderFactory.class);
 
 		Binder binder = binderFactory.getBinder(null, MessageChannel.class);
@@ -51,9 +65,70 @@ public class ErrorBindingTests {
 		applicationContext.close();
 	}
 
+	@Test
+	public void testConfigurationWithDefaultErrorHandler() {
+		ApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration
+				.getCompleteConfiguration(ErrorBindingTests.ErrorConfigurationDefault.class))
+				.web(WebApplicationType.NONE).run("--spring.cloud.stream.bindings.input.consumer.max-attempts=1", "--spring.jmx.enabled=false");
+
+		InputDestination source = context.getBean(InputDestination.class);
+		source.send(new GenericMessage<byte[]>("Hello".getBytes()));
+		source.send(new GenericMessage<byte[]>("Hello".getBytes()));
+		source.send(new GenericMessage<byte[]>("Hello".getBytes()));
+
+		ErrorConfigurationDefault errorConfiguration = context.getBean(ErrorConfigurationDefault.class);
+		assertThat(errorConfiguration.counter == 3);
+	}
+
+	@Test
+	public void testConfigurationWithCustomErrorHandler() {
+		ApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration
+				.getCompleteConfiguration(ErrorBindingTests.ErrorConfigurationWithCustomErrorHandler.class))
+				.web(WebApplicationType.NONE).run("--spring.cloud.stream.bindings.input.consumer.max-attempts=1", "--spring.jmx.enabled=false");
+
+		InputDestination source = context.getBean(InputDestination.class);
+		source.send(new GenericMessage<byte[]>("Hello".getBytes()));
+		source.send(new GenericMessage<byte[]>("Hello".getBytes()));
+		source.send(new GenericMessage<byte[]>("Hello".getBytes()));
+
+		ErrorConfigurationWithCustomErrorHandler errorConfiguration = context.getBean(ErrorConfigurationWithCustomErrorHandler.class);
+		assertThat(errorConfiguration.counter == 6);
+	}
+
 	@EnableBinding(Processor.class)
 	@EnableAutoConfiguration
 	public static class TestProcessor {
 
 	}
+
+	@EnableBinding(Processor.class)
+	@EnableAutoConfiguration
+	public static class ErrorConfigurationDefault {
+
+		private int counter;
+
+		@StreamListener(Sink.INPUT)
+		public void handle(Object value) {
+			counter++;
+			throw new RuntimeException("BOOM!");
+		}
+	}
+
+	@EnableBinding(Processor.class)
+	@EnableAutoConfiguration
+	public static class ErrorConfigurationWithCustomErrorHandler {
+
+		private int counter;
+
+		@StreamListener(Sink.INPUT)
+		public void handle(Object value) {
+			counter++;
+			throw new RuntimeException("BOOM!");
+		}
+
+		@ServiceActivator(inputChannel = "input.anonymous.errors")
+		public void error(Message<?> message) {
+			counter++;
+		}
+	}
 }