Introduce `PartitionedChannel` #8617

artembilan · 2023-05-09T20:35:59Z

Implement a PartitionedChannel as an extension of the AbstractExecutorChannel
Supply this channel with a PartitionedDispatcher which is an extension of the AbstractDispatcher
The target partition is essentially a UnicastingDispatcher with a single thead executor

* Implement a `PartitionedChannel` as an extension of the `AbstractExecutorChannel` * Supply this channel with a `PartitionedDispatcher` which is an extension of the `AbstractDispatcher` * The target partition is essentially a `UnicastingDispatcher` with a single thead executor

artembilan · 2023-05-09T20:36:32Z

If this is OK, I'll add some docs and DSL configuration.
Thanks

garyrussell

Some suggestions; otherwise LGTM.

garyrussell · 2023-05-10T13:52:02Z

...tegration-core/src/main/java/org/springframework/integration/channel/PartitionedChannel.java

+ * An {@link AbstractExecutorChannel} implementation for partitioned messages dispatching.
+ * Requires a number of partitions where each of them is backed by a dedicated thread.
+ * The {@code partitionKeyFunction} is used to determine to which partition the message
+ * has to be dispatched.
+ * <p>
+ * The actual dispatching and threading logic in implemented in the {@link PartitionedDispatcher}.
+ * <p>
+ * The default {@link ThreadFactory} is based on a bean name of this channel plus {@code -partition-thread-}.
+ * <p>
+ * The rest of the logic is similar to the {@link ExecutorChannel}, which includes:
+ * - load balancing for subscribers;
+ * - fail-over and error handling;
+ * - channel operations intercepting.


Suggested change

* An {@link AbstractExecutorChannel} implementation for partitioned messages dispatching.

* Requires a number of partitions where each of them is backed by a dedicated thread.

* The {@code partitionKeyFunction} is used to determine to which partition the message

* has to be dispatched.

* 

* The actual dispatching and threading logic in implemented in the {@link PartitionedDispatcher}.

* 

* The default {@link ThreadFactory} is based on a bean name of this channel plus {@code -partition-thread-}.

* 

* The rest of the logic is similar to the {@link ExecutorChannel}, which includes:

* - load balancing for subscribers;

* - fail-over and error handling;

* - channel operations intercepting.

* An {@link AbstractExecutorChannel} implementation for partitioned message dispatching.

* Requires a number of partitions where each of them is backed by a dedicated thread.

* The {@code partitionKeyFunction} is used to determine to which partition the message

* has to be dispatched.

* 

* The actual dispatching and threading logic in implemented in the {@link PartitionedDispatcher}.

* 

* The default {@link ThreadFactory} is based on the bean name of this channel plus {@code -partition-thread-}.

* 

* The rest of the logic is similar to the {@link ExecutorChannel}, which includes:

* - load balancing for subscribers;

* - fail-over and error handling;

* - channel operations intercepting.

...tegration-core/src/main/java/org/springframework/integration/channel/PartitionedChannel.java

garyrussell · 2023-05-10T14:01:06Z

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

+	@Override
+	public boolean dispatch(Message<?> message) {
+		int partition = this.partitionKeyFunction.apply(message).hashCode() % this.partitionCount;
+		UnicastingDispatcher partitionDispatcher = this.partitions.computeIfAbsent(partition, (__) -> newPartition());


Perhaps we should pre-populate - otherwise the thread names won't necessarily match the partition #.

garyrussell · 2023-05-10T14:06:13Z

...tion-core/src/test/java/org/springframework/integration/channel/PartitionedChannelTests.java

+					assertThat(messagesInPartition.get(0).getHeaders().get("partitionKey"))
+							.isEqualTo(messagesInPartition.get(1).getHeaders().get("partitionKey"));
+				});
+


If we match the thread name to the partition, we could assert that the records went to the right partition.

Well, it is not always true, since we don't use a partition key as is, but rather its hashCode().
So, while all the same keys goes to the same partition, it does not mean that 1 for key will always determine the partition 0.

* Add docs for `PartitionedChannel` into `channel.adoc` * Pre-populate partitions in the `PartitionedDispatcher` to ensure a thread number reflection of the partition it is used for (for default `ThreadFactory`) * Add Javadocs to `public` methods * Add Java DSL `PartitionedChannelSpec` and respective factory methods into `Channels` * Use `IntegrationMessageHeaderAccessor.CORRELATION_ID` header as a default partition key

garyrussell

Just doc polishing.

garyrussell · 2023-05-10T17:43:11Z

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

+		return partitionDispatcher.dispatch(message);
+	}
+
+	private void prePopulatedPartitionsIfAny() {


Suggested change

private void prePopulatedPartitionsIfAny() {

private synchronized void prePopulatedPartitionsIfAny() {

garyrussell · 2023-05-10T17:47:08Z

src/reference/asciidoc/channel.adoc

+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.
+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 channel can be configured as a simple bean:


Suggested change

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.

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 channel can be configured as a simple bean:

Starting with version 6.1, a `PartitionedChannel` implementation is provided.

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 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:

garyrussell · 2023-05-10T17:47:26Z

src/reference/asciidoc/channel.adoc

+----
+====
+
+The channel will have `3` partitions - dedicated threads; will use a `partitionKey` header to determine in which partition the message must be handled.


Suggested change

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 must be handled.

garyrussell · 2023-05-10T17:47:37Z

src/reference/asciidoc/channel.adoc

+----
+====
+
+The channel will have `3` partitions - dedicated threads; will use a `partitionKey` header to determine in which partition the message must be handled.


Suggested change

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.

* Fix Checkstyle violations * Mark `PartitionedDispatcher.prePopulatedPartitions()` with `synchronized`

garyrussell · 2023-05-10T18:31:09Z

...ion-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()) {


This won't work, thread 2 might proceed before thread 1 has populated all partition execs, and NPE on the dispatcher returned by the get().

How about now?
See the latest commit.

and when a custom `ThreadFactory` is set

garyrussell · 2023-05-10T18:55:47Z

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

+	public void setThreadFactory(ThreadFactory threadFactory) {
+		Assert.notNull(threadFactory, "'threadFactory' must not be null");
+		this.threadFactory = threadFactory;
+		populatedPartitions();


Need to shut down and clear existing executors.

Perhaps call shutdown() first?

Yeah... Nothing to shutdown I guess: those executors have nit been used yet.
But we indeed have to this.executors.clear() .

We don't know that; they could call this after we started processing messages.

And if not, it's a no-op; so why worry?

Doh... We cannon populate them from the ctor: some other options are required for delegate dispatchers.
So, coming back to the synchronized solution 😄

…dispatch()`

artembilan added type: enhancement in: core labels May 9, 2023

artembilan added this to the 6.1.0 milestone May 9, 2023

garyrussell suggested changes May 10, 2023

View reviewed changes

artembilan requested a review from garyrussell May 10, 2023 17:33

garyrussell suggested changes May 10, 2023

View reviewed changes

* Fix language in Javadocs

2b6d243

* Fix Checkstyle violations * Mark `PartitionedDispatcher.prePopulatedPartitions()` with `synchronized`

garyrussell suggested changes May 10, 2023

View reviewed changes

* Populate partitions from PartitionedDispatcher ctor

584f00d

and when a custom `ThreadFactory` is set

garyrussell suggested changes May 10, 2023

View reviewed changes

artembilan added 2 commits May 10, 2023 14:59

* Clear executors in populatedPartitions()

cc3a6c6

* Bring back synchronized populatedPartitions() and use it in the `…

7c6c56b

…dispatch()`

garyrussell approved these changes May 10, 2023

View reviewed changes

garyrussell merged commit 5f1e12e into spring-projects:main May 10, 2023

artembilan deleted the PartitionedChannel branch May 10, 2023 20:37

	private void prePopulatedPartitionsIfAny() {
	private synchronized void prePopulatedPartitionsIfAny() {

	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 must be handled.

Introduce PartitionedChannel #8617

Introduce PartitionedChannel #8617

Uh oh!

Conversation

artembilan commented May 9, 2023

Uh oh!

artembilan commented May 9, 2023

Uh oh!

garyrussell left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

garyrussell left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Introduce `PartitionedChannel` #8617

Introduce `PartitionedChannel` #8617