diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index c34f6322c..a0b67975f 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java @@ -38,6 +38,7 @@ import org.apache.kafka.common.serialization.ByteArraySerializer; import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.cloud.stream.binder.AbstractBinder; +import org.springframework.cloud.stream.binder.Binder; import org.springframework.cloud.stream.binder.BinderException; import org.springframework.cloud.stream.binder.BinderHeaders; import org.springframework.cloud.stream.binder.BinderPropertyKeys; @@ -92,35 +93,7 @@ import kafka.utils.ZkUtils; import scala.collection.Seq; /** - * A binder that uses Kafka as the underlying middleware. The general implementation mapping between XD concepts - * and Kafka concepts is as follows: - * A binder that uses Kafka as the underlying middleware. - * The general implementation mapping between XD concepts and Kafka concepts is as follows: - * - * - * - * - * - * - * - * - * - * - * - * - * - * - * - * - * - * - * - *
Stream definitionKafka topicKafka partitionsNotes
foo = "http | log"foo.01 partition1 producer, 1 consumer
foo = "http | log", log.count=xfoo.0x partitions1 producer, x consumers with static - * group 'springXD', achieves queue semantics
foo = "http | log", log.count=x + XD partitioningstill 1 topic 'foo.0'x partitions + use key - * computed by XD1 producer, x consumers with static group 'springXD', achieves queue semantics
foo = "http | log", log.count=x, concurrency=yfoo.0x*y partitions1 producer, x XD - * consumers, each with y threads
foo = "http | log", log.count=0, x actual log containersfoo.010(configurable) - * partitions1 producer, x XD consumers. Can't know the number of partitions beforehand, so decide a number - * that better be greater than number of containers
+ * A {@link Binder} that uses Kafka as the underlying middleware. * * @author Eric Bottard * @author Marius Bogoevici @@ -326,7 +299,6 @@ public class KafkaMessageChannelBinder extends AbstractBinder { @Override public void onInit() throws Exception { - // we instantiate the connection factory here due to https://jira.spring.io/browse/XD-2647 ZookeeperConfiguration configuration = new ZookeeperConfiguration(this.zookeeperConnect); configuration.setBufferSize(socketBufferSize); configuration.setMaxWait(defaultMaxWait); diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafkaBinderTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafkaBinderTests.java index 8dbc440e7..c07105dd7 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafkaBinderTests.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/RawModeKafkaBinderTests.java @@ -149,7 +149,6 @@ public class RawModeKafkaBinderTests extends KafkaBinderTests { .setHeader(IntegrationMessageHeaderAccessor.CORRELATION_ID, "foo") .setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, 42) .setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, 43) - .setHeader("xdReplyChannel", "bar") .build(); output.send(message2); output.send(new GenericMessage<>(new byte[]{1})); diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java index 5f2adf3fa..25c000310 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java @@ -469,11 +469,7 @@ public class RabbitMessageChannelBinder extends AbstractBinder { private Binding doRegisterConsumer(final String name, String group, MessageChannel moduleInputChannel, Queue queue, final RabbitPropertiesAccessor properties) { DefaultBinding consumerBinding = null; - // Fix for XD-2503 - // Temporarily overrides the thread context classloader with the one where the SimpleMessageListenerContainer - // is defined - // This allows for the proxying that happens while initializing the SimpleMessageListenerContainer to work - // correctly + // TODO https://github.com/spring-cloud/spring-cloud-stream/issues/401 ClassLoader originalClassloader = Thread.currentThread().getContextClassLoader(); try { ClassUtils.overrideThreadContextClassLoader(SimpleMessageListenerContainer.class.getClassLoader()); diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-test/README.md b/spring-cloud-stream-binders/spring-cloud-stream-binder-test/README.md deleted file mode 100644 index b49eeecc2..000000000 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-test/README.md +++ /dev/null @@ -1,2 +0,0 @@ -Spring XD Test Support -====================== diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/PartitionCapableBinderTests.java b/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/PartitionCapableBinderTests.java index 3e18203d6..5a080d729 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/PartitionCapableBinderTests.java +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-test/src/main/java/org/springframework/cloud/stream/binder/PartitionCapableBinderTests.java @@ -353,7 +353,7 @@ abstract public class PartitionCapableBinderTests extends BrokerBinderTests { /** * Implementations should return whether the binder under test uses "explicit" routing (e.g. Rabbit) - * whereby XD is responsible for assigning a partition and knows which exact consumer will receive the + * whereby Spring Cloud Stream is responsible for assigning a partition and knows which exact consumer will receive the * message (i.e. honor "partitionIndex") or "implicit" routing (e.g. Kafka) whereby the only guarantee * is that messages will be spread, but we don't control exactly which consumer gets which message. */ diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderHeaders.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderHeaders.java index 0d991b3ad..0d31fb22b 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderHeaders.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderHeaders.java @@ -26,10 +26,7 @@ import org.springframework.messaging.MessageHeaders; * @author David Turanski */ public final class BinderHeaders { - - /* - * no xd prefix for backwards compatibility - */ + public static final String BINDER_ORIGINAL_CONTENT_TYPE = "originalContentType"; /**