DefaultMessageEndpoint now sets the reply Message's correlationId to the original Message's id by default (INT-130).

spring-integration-core/src/test/java/org/springframework/integration/endpoint/DefaultMessageEndpointTests.java

@@ -17,6 +17,7 @@
 package org.springframework.integration.endpoint;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
@@ -521,6 +522,41 @@ public class DefaultMessageEndpointTests {
 		assertEquals(message2, ((MessageDeliveryException) error).getUndeliveredMessage());
 	}
 
+	@Test
+	public void testCorrelationId() {
+		SimpleChannel replyChannel = new SimpleChannel(1);
+		DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(new MessageHandler() {
+			public Message<?> handle(Message<?> message) {
+				return message;
+			}
+		});
+		endpoint.start();
+		Message<?> message = new StringMessage("test");
+		message.getHeader().setReturnAddress(replyChannel);
+		endpoint.handle(message);
+		Message<?> reply = replyChannel.receive(500);
+		assertEquals(message.getId(), reply.getHeader().getCorrelationId());
+	}
+
+	@Test
+	public void testCorrelationIdSetByHandlerTakesPrecedence() {
+		SimpleChannel replyChannel = new SimpleChannel(1);
+		DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(new MessageHandler() {
+			public Message<?> handle(Message<?> message) {
+				message.getHeader().setCorrelationId("ABC-123");
+				return message;
+			}
+		});
+		endpoint.start();
+		Message<?> message = new StringMessage("test");
+		message.getHeader().setReturnAddress(replyChannel);
+		endpoint.handle(message);
+		Message<?> reply = replyChannel.receive(500);
+		Object correlationId = reply.getHeader().getCorrelationId();
+		assertFalse(message.getId().equals(correlationId));
+		assertEquals("ABC-123", correlationId);
+	}
+
 
 	private static ExecutorService createExecutor() {
 		return new ThreadPoolExecutor(1, 1, 60, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());