Add mechanism for pro-active routing connection management #802
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
For many types of jobs, connection changes will be far less frequent then `route` calls. Therefore, it can be quite expensive to re-calculate the underlying connection routing on every call to `route`. This update introduces a new router type `ProactiveRouter`. These routers effectively subscribe to changes in the connections and are responsible for maintaining the state of the connections when routing. For example on the `ConsistentHashingRouter` this means that they can just add or remove a connection from the map instead of re-calculating everything from scratch. I disabled this feature by default and required it to be explicitly turned on. Additionally, I add some hooks to the RouterFactory (should be the only breaking change) so that we can hook in our own custom router factories for these connections.
timmartin-stripe
changed the title
| import java.util.concurrent.locks.ReadWriteLock; | ||
| import java.util.concurrent.locks.ReentrantReadWriteLock; | ||
|
|
||
| public class ProactiveConsistentHashingRouter<K, V> implements ProactiveRouter<KeyValuePair<K, V>> { |
Contributor
Author
There was a problem hiding this comment.
Basically the same as ConsistentHashingRouter except that does not recalculate the whole thing. It just makes the updates to add/remove a connection.
| <T> Router<T> scalarStageToStageRouter(String name, final Func1<T, byte[]> toBytes); | ||
|
|
||
| default <T> ProactiveRouter<T> scalarStageToStageProactiveRouter(String name, final Func1<T, byte[]> toBytes) { | ||
| return new ProactiveRoundRobinRouter<>(name, toBytes); |
Contributor
Author
There was a problem hiding this comment.
defaults for backwards compatibility.
| public static <K, V> Router<KeyValuePair<K, V>> consistentHashingLegacyTcpProtocol(String name, | ||
| final Func1<K, byte[]> keyEncoder, | ||
| final Func1<V, byte[]> valueEncoder) { | ||
| return new ConsistentHashingRouter<K, V>(name, new Func1<KeyValuePair<K, V>, byte[]>() { |
Contributor
Author
There was a problem hiding this comment.
Just moved to a static method on the factory to avoid circular dependencies.
…nection-management
timmartin-stripe
requested review from
Andyz26,
calvin681,
dtrager02,
fdc-ntflx,
hellolittlej and
james-lubin
as code owners
Andyz26
reviewed
Oct 27, 2025
| private final Optional<ProactiveRouter<T>> router; | ||
|
|
||
| public ConnectionGroup(String groupId) { | ||
| public ConnectionGroup(String groupId, Optional<ProactiveRouter<T>> router) { |
| public ProactiveRoundRobinRouter(String name, Func1<T, byte[]> encoder) { | ||
| this.encoder = encoder; | ||
| metrics = new Metrics.Builder() | ||
| .name("Router_" + name) |
Collaborator
There was a problem hiding this comment.
prefix metrix name with "ProactiveRoundRobin"?
Contributor
Author
There was a problem hiding this comment.
Added a tag
| this.connectionRepetitionOnRing = ringRepetitionPerConnection; | ||
| this.encoder = dataEncoder; | ||
| metrics = new Metrics.Builder() | ||
| .name("Router_" + name) |
Collaborator
There was a problem hiding this comment.
prefix metrix name with "ProactiveRoundRobin"?
Contributor
Author
|
@Andyz26 Thanks for the review! Made the requested updates. |
timmartin-stripe
requested a deployment
to
Integrate Pull Request
GitHub Actions
Waiting
Andyz26
approved these changes
Oct 28, 2025
| return this; | ||
| } | ||
|
|
||
| public Config<K, T, R> shouldUseProactiveRouter(boolean useProactiveRouter) { |
Collaborator
There was a problem hiding this comment.
shall these names be consistent across stage config? e.g withProactiveRouter in keytokey.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Context
For many types of jobs, connection changes will be far less frequent
then
routecalls. Therefore, it can be quite expensive tore-calculate the underlying connection routing on every call to
route.In some of our jobs at Stripe, the ConsistentHashingRouter was a limiting
factor in how many events/second/host we could process.
This update introduces a new router type
ProactiveRouter. Theserouters effectively subscribe to changes in the connections and are
responsible for maintaining the state of the connections when routing.
For example on the
ConsistentHashingRouterthis means that they canjust add or remove a connection from the map instead of re-calculating
everything from scratch.
I disabled this feature by default and required it to be explicitly
turned on.
Checklist
./gradlew buildcompiles code correctly./gradlew testpasses all tests