XD-3621 Kafka bus: Pass header propagation config

Allow the headers passed by the kafka bus to be configured.

Update kafka-bus.xml

Polishing

Polishing on Kafka tests

Author: Marius Bogoevici

config/servers.yml

@@ -59,7 +59,7 @@
 
 #    redis:
 #      headers:
-            # comman-delimited list of additional (string-valued) header names to transport
+            # comma-delimited list of additional header names to transport
 #      default:
             # default bus properties, if not specified at the module level
 #        backOffInitialInterval:    1000
@@ -71,6 +71,8 @@
 #      brokers:                                 localhost:9092
 #      zkAddress:                               localhost:2181
 #      mode:                                    embeddedHeaders
+#      headers:
+            # comma-delimited list of additional header names to transport
 #      socketBufferSize:                        2097152
 #      offsetStoreTopic:                        SpringXdOffsets
 #      offsetStoreSegmentSize:                  25000000

spring-xd-dirt/src/main/resources/application.yml

@@ -142,7 +142,7 @@ xd:
         txSize:                    1
     redis:
       headers:
-            # comman-delimited list of additional (string-valued) header names to transport
+            # comma-delimited list of additional header names to transport
       default:
             # default bus properties, if not specified at the module level
         backOffInitialInterval:    1000
@@ -154,6 +154,8 @@ xd:
       brokers:                                 localhost:9092
       zkAddress:                               localhost:2181
       mode:                                    embeddedHeaders
+      headers:
+            # comma-delimited list of additional header names to transport
       socketBufferSize:                        2097152
       offsetStoreTopic:                        SpringXdOffsets
       offsetStoreSegmentSize:                  25000000

spring-xd-dirt/src/test/java/org/springframework/xd/dirt/integration/bus/kafka/KafkaMessageBusTests.java

@@ -20,8 +20,10 @@
 import static org.hamcrest.collection.IsCollectionWithSize.hasSize;
 import static org.hamcrest.core.IsEqual.equalTo;
 import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
 
 import java.util.Arrays;
 import java.util.Collection;
@@ -32,10 +34,9 @@
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.TimeUnit;
 
-import kafka.api.OffsetRequest;
 import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.ProducerConfig;
-import org.hamcrest.CoreMatchers;
+import org.hamcrest.collection.IsMapContaining;
 import org.junit.Ignore;
 import org.junit.Rule;
 import org.junit.Test;
@@ -50,15 +51,19 @@
 import org.springframework.integration.kafka.listener.KafkaMessageListenerContainer;
 import org.springframework.integration.kafka.listener.MessageListener;
 import org.springframework.integration.kafka.support.ProducerConfiguration;
+import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.xd.dirt.integration.bus.Binding;
 import org.springframework.xd.dirt.integration.bus.BusProperties;
 import org.springframework.xd.dirt.integration.bus.MessageBus;
 import org.springframework.xd.dirt.integration.bus.PartitionCapableBusTests;
+import org.springframework.xd.dirt.integration.bus.XdHeaders;
 import org.springframework.xd.dirt.integration.kafka.KafkaMessageBus;
 import org.springframework.xd.test.kafka.KafkaTestSupport;
 
+import kafka.api.OffsetRequest;
+
 
 /**
  * Integration tests for the {@link KafkaMessageBus}.
@@ -368,6 +373,35 @@ public void testKafkaSpecificProducerPropertiesAreSet() {
 		messageBus.unbindProducers("foo" + uniqueBindingId);
 	}
 
+	@Test
+	public void testMoreHeaders() throws Exception {
+		KafkaTestMessageBus bus =
+				new KafkaTestMessageBus(kafkaTestSupport, getCodec(), KafkaMessageBus.Mode.embeddedHeaders,
+						"propagatedHeader");
+		Collection<String> headers = Arrays.asList(TestUtils.getPropertyValue(bus.getCoreMessageBus(), "headersToMap",
+				String[].class));
+		assertTrue(headers.contains("propagatedHeader"));
+		assertEquals(XdHeaders.STANDARD_HEADERS.length + 1, headers.size());
+		DirectChannel moduleOutputChannel = new DirectChannel();
+		QueueChannel moduleInputChannel = new QueueChannel();
+		long uniqueBindingId = System.currentTimeMillis();
+		Properties emptyProperties = new Properties();
+		bus.bindProducer("foo" + uniqueBindingId + ".0", moduleOutputChannel, emptyProperties);
+		bus.bindConsumer("foo" + uniqueBindingId + ".0", moduleInputChannel, emptyProperties);
+		Message<?> message = org.springframework.integration.support.MessageBuilder.
+				withPayload("payload").setHeader("propagatedHeader","propagatedValue").build();
+		// Let the consumer actually bind to the producer before sending a msg
+		busBindUnbindLatency();
+		moduleOutputChannel.send(message);
+		Message<?> inbound = moduleInputChannel.receive(2000);
+		assertThat(inbound.getHeaders(), IsMapContaining.hasKey("propagatedHeader"));
+		assertThat((String)inbound.getHeaders().get("propagatedHeader"), equalTo("propagatedValue"));
+		bus.unbindProducers("foo" + uniqueBindingId + ".0");
+		bus.unbindConsumers("foo" + uniqueBindingId + ".0");
+		bus.cleanup();
+	}
+
+
 	@Test
 	@Ignore("Kafka message bus does not support direct binding")
 	@Override

spring-xd-dirt/src/test/java/org/springframework/xd/dirt/integration/bus/kafka/KafkaTestMessageBus.java

@@ -43,14 +43,14 @@ public KafkaTestMessageBus(KafkaTestSupport kafkaTestSupport) {
 
 
 	public KafkaTestMessageBus(KafkaTestSupport kafkaTestSupport, Codec codec,
-			KafkaMessageBus.Mode mode) {
+			KafkaMessageBus.Mode mode, String... headers) {
 
 		try {
 			ZookeeperConnect zookeeperConnect = new ZookeeperConnect();
 			zookeeperConnect.setZkConnect(kafkaTestSupport.getZkConnectString());
 			KafkaMessageBus messageBus = new KafkaMessageBus(zookeeperConnect,
 					kafkaTestSupport.getBrokerAddress(),
-					kafkaTestSupport.getZkConnectString(), codec);
+					kafkaTestSupport.getZkConnectString(), codec, headers);
 			messageBus.setDefaultBatchingEnabled(false);
 			messageBus.setMode(mode);
 			messageBus.afterPropertiesSet();

spring-xd-dirt/src/test/java/org/springframework/xd/dirt/integration/bus/redis/RedisMessageBusTests.java

@@ -37,14 +37,14 @@
 import org.junit.Rule;
 import org.junit.Test;
 
-import org.springframework.amqp.utils.test.TestUtils;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.data.redis.serializer.StringRedisSerializer;
 import org.springframework.expression.Expression;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.endpoint.AbstractEndpoint;
 import org.springframework.integration.redis.inbound.RedisQueueMessageDrivenEndpoint;
+import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.retry.support.RetryTemplate;
 import org.springframework.xd.dirt.integration.bus.Binding;

spring-xd-messagebus-kafka/src/main/resources/META-INF/spring-xd/bus/kafka-bus.xml

@@ -16,7 +16,7 @@
 		<constructor-arg value="${xd.messagebus.kafka.brokers}"/>
 		<constructor-arg value="${xd.messagebus.kafka.zkAddress}"/>
 		<constructor-arg ref="codec"/>
-		<constructor-arg value="#{new String[0]}"/>
+		<constructor-arg value="${xd.messagebus.kafka.headers:}" />
 
 		<property name="mode" value="${xd.messagebus.kafka.mode}"/>
 

spring-xd-yarn/site/config/servers.yml

@@ -166,7 +166,7 @@ xd:
 
 #    redis:
 #      headers:
-            # comman-delimited list of additional (string-valued) header names to transport
+            # comma-delimited list of additional header names to transport
 #      default:
             # default bus properties, if not specified at the module level
 #        backOffInitialInterval:    1000
@@ -179,6 +179,8 @@ xd:
 #      brokers:                                 localhost:9092
 #      zkAddress:                               localhost:2181
 #      mode:                                    embeddedHeaders
+#      headers:
+            # comma-delimited list of additional header names to transport
 #      socketBufferSize:                        2097152
 #      offsetStoreTopic:                        SpringXdOffsets
 #      offsetStoreSegmentSize:                  25000000