Introduce PartitionedChannel

spring-integration-core/src/main/java/org/springframework/integration/dispatcher/PartitionedDispatcher.java

@@ -145,13 +145,15 @@ public void shutdown() {
 
 	@Override
 	public boolean dispatch(Message<?> message) {
-		prePopulatedPartitionsIfAny();
+		if (this.partitions.isEmpty()) {
+			prePopulatedPartitions();
+		}
 		int partition = Math.abs(this.partitionKeyFunction.apply(message).hashCode()) % this.partitionCount;
 		UnicastingDispatcher partitionDispatcher = this.partitions.get(partition);
 		return partitionDispatcher.dispatch(message);
 	}
 
-	private void prePopulatedPartitionsIfAny() {
+	private synchronized void prePopulatedPartitions() {
 		if (this.partitions.isEmpty()) {
 			for (int i = 0; i < this.partitionCount; i++) {
 				this.partitions.put(i, newPartition());

spring-integration-core/src/main/java/org/springframework/integration/dsl/PartitionedChannelSpec.java

@@ -19,11 +19,7 @@
 import java.util.concurrent.ThreadFactory;
 import java.util.function.Function;
 
-import org.springframework.integration.channel.AbstractExecutorChannel;
-import org.springframework.integration.channel.AbstractMessageChannel;
-import org.springframework.integration.channel.ExecutorChannel;
 import org.springframework.integration.channel.PartitionedChannel;
-import org.springframework.integration.dispatcher.PartitionedDispatcher;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 
@@ -61,10 +57,10 @@ public PartitionedChannelSpec threadFactory(ThreadFactory threadFactory) {
 	@Override
 	protected PartitionedChannel doGet() {
 		if (this.partitionKeyFunction != null) {
-			this.channel = new PartitionedChannel(partitionCount, this.partitionKeyFunction);
+			this.channel = new PartitionedChannel(this.partitionCount, this.partitionKeyFunction);
 		}
 		else {
-			this.channel = new PartitionedChannel(partitionCount);
+			this.channel = new PartitionedChannel(this.partitionCount);
 		}
 		this.channel.setLoadBalancingStrategy(this.loadBalancingStrategy);
 		if (this.failover != null) {

src/reference/asciidoc/channel.adoc

@@ -207,11 +207,11 @@ Since that situation would only occur in a non-predictable way, you should not r
 ===== `PartitionedChannel`
 
 Starting with version 6.1, a `PartitionedChannel` implementation is provided.
-This is an extension of `AbstractExecutorChannel` and represents a point-to-point dispatching logic where an actual consumption happens on a specific thread determined by the partition key evaluated from a message sent to this channel.
-This channel is similar to the `ExecutorChannel` mentioned above, but with a difference that messages with the same partition key are always handled in the same thread preserving ordering.
+This is an extension of `AbstractExecutorChannel` and represents point-to-point dispatching logic where the actual consumption is processed on a specific thread, determined by the partition key evaluated from a message sent to this channel.
+This channel is similar to the `ExecutorChannel` mentioned above, but with the difference that messages with the same partition key are always handled in the same thread, preserving ordering.
 It does not require an external `TaskExecutor`, but can be configured with a custom `ThreadFactory` (e.g. `Thread.ofVirtual().name("partition-", 0).factory()`).
-This factory then used to populate single-thread executors into `MessageDispatcher` delegates per partition.
-By default, the `IntegrationMessageHeaderAccessor.CORRELATION_ID` message header is used for partition key.
+This factory is used to populate single-thread executors into a `MessageDispatcher` delegate, per partition.
+By default, the `IntegrationMessageHeaderAccessor.CORRELATION_ID` message header is used as the partition key.
 This channel can be configured as a simple bean:
 
 ====
@@ -224,7 +224,7 @@ PartitionedChannel somePartitionedChannel() {
 ----
 ====
 
-The channel will have `3` partitions - dedicated threads; will use a `partitionKey` header to determine in which partition the message must be handled.
+The channel will have `3` partitions - dedicated threads; will use the `partitionKey` header to determine in which partition the message will be handled.
 See `PartitionedChannel` class Javadocs for more information.
 
 [[flux-message-channel]]