Skip to content

Commit 5d9c2c1

Browse files
RomehRobWin
Romeh
authored and
RobWin
committed
Issue ReactiveX#733: Added bulkhead/ThreadPoolBulhead spring customizer (ReactiveX#797)
1 parent e32438c commit 5d9c2c1

File tree

18 files changed

+316
-71
lines changed

18 files changed

+316
-71
lines changed

‎resilience4j-framework-common/src/main/java/io/github/resilience4j/common/CompositeCustomizer.java

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,18 @@
1+
/*
2+
* Copyright Mahmoud Romeh
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
116
package io.github.resilience4j.common;
217

318
import java.util.HashMap;

‎resilience4j-framework-common/src/main/java/io/github/resilience4j/common/CustomizerWithName.java

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,18 @@
1+
/*
2+
* Copyright Mahmoud Romeh
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
116
package io.github.resilience4j.common;
217

318
/**

‎resilience4j-framework-common/src/main/java/io/github/resilience4j/common/bulkhead/configuration/BulkheadConfigCustomizer.java

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,18 @@
1+
package io.github.resilience4j.common.bulkhead.configuration;
2+
3+
import io.github.resilience4j.bulkhead.BulkheadConfig;
4+
import io.github.resilience4j.common.CustomizerWithName;
5+
6+
/**
7+
* Enable customization bulkhead configuration builders programmatically.
8+
*/
9+
public interface BulkheadConfigCustomizer extends CustomizerWithName {
10+
11+
/**
12+
* Customize BulkheadConfig configuration builder.
13+
*
14+
* @param configBuilder to be customized
15+
*/
16+
void customize(BulkheadConfig.Builder configBuilder);
17+
18+
}

‎resilience4j-framework-common/src/main/java/io/github/resilience4j/common/bulkhead/configuration/BulkheadConfigurationProperties.java

Lines changed: 20 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717

1818
import io.github.resilience4j.bulkhead.BulkheadConfig;
1919
import io.github.resilience4j.common.CommonProperties;
20+
import io.github.resilience4j.common.CompositeCustomizer;
2021
import io.github.resilience4j.common.utils.ConfigUtils;
2122
import io.github.resilience4j.core.ConfigurationNotFoundException;
2223
import io.github.resilience4j.core.StringUtils;
@@ -32,32 +33,44 @@ public class BulkheadConfigurationProperties extends CommonProperties {
3233
private Map<String, InstanceProperties> instances = new HashMap<>();
3334
private Map<String, InstanceProperties> configs = new HashMap<>();
3435

35-
public BulkheadConfig createBulkheadConfig(InstanceProperties instanceProperties) {
36+
public BulkheadConfig createBulkheadConfig(InstanceProperties instanceProperties,
37+
CompositeCustomizer<BulkheadConfigCustomizer> compositeBulkheadCustomizer,
38+
String instanceName) {
3639
if (StringUtils.isNotEmpty(instanceProperties.getBaseConfig())) {
3740
InstanceProperties baseProperties = configs.get(instanceProperties.getBaseConfig());
3841
if (baseProperties == null) {
3942
throw new ConfigurationNotFoundException(instanceProperties.getBaseConfig());
4043
}
41-
return buildConfigFromBaseConfig(baseProperties, instanceProperties);
44+
return buildConfigFromBaseConfig(baseProperties, instanceProperties,
45+
compositeBulkheadCustomizer, instanceName);
4246
}
43-
return buildBulkheadConfig(BulkheadConfig.custom(), instanceProperties);
47+
return buildBulkheadConfig(BulkheadConfig.custom(), instanceProperties,
48+
compositeBulkheadCustomizer, instanceName);
4449
}
4550

4651
private BulkheadConfig buildConfigFromBaseConfig(InstanceProperties baseProperties,
47-
InstanceProperties instanceProperties) {
52+
InstanceProperties instanceProperties,
53+
CompositeCustomizer<BulkheadConfigCustomizer> compositeBulkheadCustomizer,
54+
String instanceName) {
4855
ConfigUtils.mergePropertiesIfAny(baseProperties, instanceProperties);
49-
BulkheadConfig baseConfig = buildBulkheadConfig(BulkheadConfig.custom(), baseProperties);
50-
return buildBulkheadConfig(BulkheadConfig.from(baseConfig), instanceProperties);
56+
BulkheadConfig baseConfig = buildBulkheadConfig(BulkheadConfig.custom(), baseProperties,
57+
compositeBulkheadCustomizer, instanceName);
58+
return buildBulkheadConfig(BulkheadConfig.from(baseConfig), instanceProperties,
59+
compositeBulkheadCustomizer, instanceName);
5160
}
5261

5362
private BulkheadConfig buildBulkheadConfig(BulkheadConfig.Builder builder,
54-
InstanceProperties instanceProperties) {
63+
InstanceProperties instanceProperties,
64+
CompositeCustomizer<BulkheadConfigCustomizer> compositeBulkheadCustomizer,
65+
String instanceName) {
5566
if (instanceProperties.getMaxConcurrentCalls() != null) {
5667
builder.maxConcurrentCalls(instanceProperties.getMaxConcurrentCalls());
5768
}
5869
if (instanceProperties.getMaxWaitDuration() != null) {
5970
builder.maxWaitDuration(instanceProperties.getMaxWaitDuration());
6071
}
72+
compositeBulkheadCustomizer.getCustomizer(instanceName)
73+
.ifPresent(bulkheadConfigCustomizer -> bulkheadConfigCustomizer.customize(builder));
6174
return builder.build();
6275
}
6376

‎resilience4j-framework-common/src/main/java/io/github/resilience4j/common/bulkhead/configuration/ThreadPoolBulkheadConfigCustomizer.java

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,18 @@
1+
package io.github.resilience4j.common.bulkhead.configuration;
2+
3+
import io.github.resilience4j.bulkhead.ThreadPoolBulkheadConfig;
4+
import io.github.resilience4j.common.CustomizerWithName;
5+
6+
/**
7+
* Enable customization thread pool bulkhead configuration builders programmatically.
8+
*/
9+
public interface ThreadPoolBulkheadConfigCustomizer extends CustomizerWithName {
10+
11+
/**
12+
* Customize ThreadPoolBulkheadConfig configuration builder.
13+
*
14+
* @param configBuilder to be customized
15+
*/
16+
void customize(ThreadPoolBulkheadConfig.Builder configBuilder);
17+
18+
}

‎resilience4j-framework-common/src/main/java/io/github/resilience4j/common/bulkhead/configuration/ThreadPoolBulkheadConfigurationProperties.java

Lines changed: 24 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818
import io.github.resilience4j.bulkhead.Bulkhead;
1919
import io.github.resilience4j.bulkhead.ThreadPoolBulkheadConfig;
2020
import io.github.resilience4j.common.CommonProperties;
21+
import io.github.resilience4j.common.CompositeCustomizer;
2122
import io.github.resilience4j.core.ConfigurationNotFoundException;
2223
import io.github.resilience4j.core.StringUtils;
2324
import io.github.resilience4j.core.lang.Nullable;
@@ -53,33 +54,44 @@ public InstanceProperties getBackendProperties(String backend) {
5354
}
5455

5556
// Thread pool bulkhead section
56-
public ThreadPoolBulkheadConfig createThreadPoolBulkheadConfig(String backend) {
57-
return createThreadPoolBulkheadConfig(getBackendProperties(backend));
57+
public ThreadPoolBulkheadConfig createThreadPoolBulkheadConfig(String backend,
58+
CompositeCustomizer<ThreadPoolBulkheadConfigCustomizer> compositeThreadPoolBulkheadCustomizer) {
59+
return createThreadPoolBulkheadConfig(getBackendProperties(backend),
60+
compositeThreadPoolBulkheadCustomizer, backend);
5861
}
5962

6063
public ThreadPoolBulkheadConfig createThreadPoolBulkheadConfig(
61-
InstanceProperties instanceProperties) {
64+
InstanceProperties instanceProperties,
65+
CompositeCustomizer<ThreadPoolBulkheadConfigCustomizer> compositeThreadPoolBulkheadCustomizer,
66+
String instanceName) {
6267
if (instanceProperties != null && StringUtils
6368
.isNotEmpty(instanceProperties.getBaseConfig())) {
6469
InstanceProperties baseProperties = configs.get(instanceProperties.getBaseConfig());
6570
if (baseProperties == null) {
6671
throw new ConfigurationNotFoundException(instanceProperties.getBaseConfig());
6772
}
68-
return buildThreadPoolConfigFromBaseConfig(baseProperties, instanceProperties);
73+
return buildThreadPoolConfigFromBaseConfig(baseProperties, instanceProperties,
74+
compositeThreadPoolBulkheadCustomizer, instanceName);
6975
}
70-
return buildThreadPoolBulkheadConfig(ThreadPoolBulkheadConfig.custom(), instanceProperties);
76+
return buildThreadPoolBulkheadConfig(ThreadPoolBulkheadConfig.custom(), instanceProperties,
77+
compositeThreadPoolBulkheadCustomizer, instanceName);
7178
}
7279

7380
private ThreadPoolBulkheadConfig buildThreadPoolConfigFromBaseConfig(
74-
InstanceProperties baseProperties, InstanceProperties instanceProperties) {
81+
InstanceProperties baseProperties, InstanceProperties instanceProperties,
82+
CompositeCustomizer<ThreadPoolBulkheadConfigCustomizer> compositeThreadPoolBulkheadCustomizer,
83+
String instanceName) {
7584
ThreadPoolBulkheadConfig baseConfig = buildThreadPoolBulkheadConfig(
76-
ThreadPoolBulkheadConfig.custom(), baseProperties);
85+
ThreadPoolBulkheadConfig.custom(), baseProperties,
86+
compositeThreadPoolBulkheadCustomizer, instanceName);
7787
return buildThreadPoolBulkheadConfig(ThreadPoolBulkheadConfig.from(baseConfig),
78-
instanceProperties);
88+
instanceProperties, compositeThreadPoolBulkheadCustomizer, instanceName);
7989
}
8090

8191
public ThreadPoolBulkheadConfig buildThreadPoolBulkheadConfig(
82-
ThreadPoolBulkheadConfig.Builder builder, InstanceProperties properties) {
92+
ThreadPoolBulkheadConfig.Builder builder, InstanceProperties properties,
93+
CompositeCustomizer<ThreadPoolBulkheadConfigCustomizer> compositeThreadPoolBulkheadCustomizer,
94+
String instanceName) {
8395
if (properties == null) {
8496
return ThreadPoolBulkheadConfig.custom().build();
8597
}
@@ -99,6 +111,9 @@ public ThreadPoolBulkheadConfig buildThreadPoolBulkheadConfig(
99111
if (properties.getWritableStackTraceEnabled() != null) {
100112
builder.writableStackTraceEnabled(properties.getWritableStackTraceEnabled());
101113
}
114+
compositeThreadPoolBulkheadCustomizer.getCustomizer(instanceName).ifPresent(
115+
threadPoolBulkheadConfigCustomizer -> threadPoolBulkheadConfigCustomizer
116+
.customize(builder));
102117

103118
return builder.build();
104119
}

‎resilience4j-framework-common/src/test/java/io/github/resilience4j/common/bulkhead/configuration/BulkheadConfigurationPropertiesTest.java

Lines changed: 32 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -17,12 +17,12 @@
1717

1818
import io.github.resilience4j.bulkhead.BulkheadConfig;
1919
import io.github.resilience4j.bulkhead.ThreadPoolBulkheadConfig;
20+
import io.github.resilience4j.common.CompositeCustomizer;
2021
import io.github.resilience4j.core.ConfigurationNotFoundException;
21-
import io.vavr.Tuple;
22-
import io.vavr.Tuple2;
2322
import org.junit.Test;
2423

2524
import java.time.Duration;
25+
import java.util.Collections;
2626
import java.util.HashMap;
2727
import java.util.Map;
2828

@@ -46,21 +46,21 @@ public void tesFixedThreadPoolBulkHeadProperties() {
4646
ThreadPoolBulkheadConfigurationProperties bulkheadConfigurationProperties = new ThreadPoolBulkheadConfigurationProperties();
4747
bulkheadConfigurationProperties.getBackends().put("backend1", backendProperties1);
4848
bulkheadConfigurationProperties.getBackends().put("backend2", backendProperties2);
49-
Map<String,String> tags=new HashMap<>();
50-
tags.put("testKey1","testKet2");
49+
Map<String, String> tags = new HashMap<>();
50+
tags.put("testKey1", "testKet2");
5151
bulkheadConfigurationProperties.setTags(tags);
5252

5353
//Then
5454
assertThat(bulkheadConfigurationProperties.getTags()).isNotEmpty();
5555
assertThat(bulkheadConfigurationProperties.getBackends().size()).isEqualTo(2);
5656
assertThat(bulkheadConfigurationProperties.getInstances().size()).isEqualTo(2);
5757
ThreadPoolBulkheadConfig bulkhead1 = bulkheadConfigurationProperties
58-
.createThreadPoolBulkheadConfig("backend1");
58+
.createThreadPoolBulkheadConfig("backend1", compositeThreadPoolBulkheadCustomizer());
5959
assertThat(bulkhead1).isNotNull();
6060
assertThat(bulkhead1.getCoreThreadPoolSize()).isEqualTo(1);
6161

6262
ThreadPoolBulkheadConfig bulkhead2 = bulkheadConfigurationProperties
63-
.createThreadPoolBulkheadConfig("backend2");
63+
.createThreadPoolBulkheadConfig("backend2", compositeThreadPoolBulkheadCustomizer());
6464
assertThat(bulkhead2).isNotNull();
6565
assertThat(bulkhead2.getCoreThreadPoolSize()).isEqualTo(2);
6666

@@ -102,19 +102,22 @@ public void testCreateThreadPoolBulkHeadPropertiesWithSharedConfigs() {
102102
assertThat(bulkheadConfigurationProperties.getBackends().size()).isEqualTo(2);
103103
// Should get default config and core number
104104
ThreadPoolBulkheadConfig bulkhead1 = bulkheadConfigurationProperties
105-
.createThreadPoolBulkheadConfig("backendWithDefaultConfig");
105+
.createThreadPoolBulkheadConfig("backendWithDefaultConfig",
106+
compositeThreadPoolBulkheadCustomizer());
106107
assertThat(bulkhead1).isNotNull();
107108
assertThat(bulkhead1.getCoreThreadPoolSize()).isEqualTo(3);
108109
assertThat(bulkhead1.getQueueCapacity()).isEqualTo(1);
109110
// Should get shared config and overwrite core number
110111
ThreadPoolBulkheadConfig bulkhead2 = bulkheadConfigurationProperties
111-
.createThreadPoolBulkheadConfig("backendWithSharedConfig");
112+
.createThreadPoolBulkheadConfig("backendWithSharedConfig",
113+
compositeThreadPoolBulkheadCustomizer());
112114
assertThat(bulkhead2).isNotNull();
113115
assertThat(bulkhead2.getCoreThreadPoolSize()).isEqualTo(4);
114116
assertThat(bulkhead2.getQueueCapacity()).isEqualTo(2);
115117
// Unknown backend should get default config of Registry
116118
ThreadPoolBulkheadConfig bulkhead3 = bulkheadConfigurationProperties
117-
.createThreadPoolBulkheadConfig("unknownBackend");
119+
.createThreadPoolBulkheadConfig("unknownBackend",
120+
compositeThreadPoolBulkheadCustomizer());
118121
assertThat(bulkhead3).isNotNull();
119122
assertThat(bulkhead3.getCoreThreadPoolSize())
120123
.isEqualTo(ThreadPoolBulkheadConfig.DEFAULT_CORE_THREAD_POOL_SIZE);
@@ -140,19 +143,19 @@ public void testBulkHeadProperties() {
140143
BulkheadConfigurationProperties bulkheadConfigurationProperties = new BulkheadConfigurationProperties();
141144
bulkheadConfigurationProperties.getInstances().put("backend1", instanceProperties1);
142145
bulkheadConfigurationProperties.getInstances().put("backend2", instanceProperties2);
143-
Map<String,String> globalTags=new HashMap<>();
144-
globalTags.put("testKey1","testKet2");
146+
Map<String, String> globalTags = new HashMap<>();
147+
globalTags.put("testKey1", "testKet2");
145148
bulkheadConfigurationProperties.setTags(globalTags);
146149
//Then
147150
assertThat(bulkheadConfigurationProperties.getInstances().size()).isEqualTo(2);
148151
assertThat(bulkheadConfigurationProperties.getTags()).isNotEmpty();
149152
BulkheadConfig bulkhead1 = bulkheadConfigurationProperties
150-
.createBulkheadConfig(instanceProperties1);
153+
.createBulkheadConfig(instanceProperties1, compositeBulkheadCustomizer(), "backend1");
151154
assertThat(bulkhead1).isNotNull();
152155
assertThat(bulkhead1.getMaxConcurrentCalls()).isEqualTo(3);
153156

154157
BulkheadConfig bulkhead2 = bulkheadConfigurationProperties
155-
.createBulkheadConfig(instanceProperties2);
158+
.createBulkheadConfig(instanceProperties2, compositeBulkheadCustomizer(), "backend2");
156159
assertThat(bulkhead2).isNotNull();
157160
assertThat(bulkhead2.getMaxConcurrentCalls()).isEqualTo(2);
158161

@@ -196,21 +199,24 @@ public void testCreateBulkHeadPropertiesWithSharedConfigs() {
196199

197200
// Should get default config and overwrite max calls and wait time
198201
BulkheadConfig bulkhead1 = bulkheadConfigurationProperties
199-
.createBulkheadConfig(backendWithDefaultConfig);
202+
.createBulkheadConfig(backendWithDefaultConfig, compositeBulkheadCustomizer(),
203+
"backendWithDefaultConfig");
200204
assertThat(bulkhead1).isNotNull();
201205
assertThat(bulkhead1.getMaxConcurrentCalls()).isEqualTo(3);
202206
assertThat(bulkhead1.getMaxWaitDuration().toMillis()).isEqualTo(200L);
203207

204208
// Should get shared config and overwrite wait time
205209
BulkheadConfig bulkhead2 = bulkheadConfigurationProperties
206-
.createBulkheadConfig(backendWithSharedConfig);
210+
.createBulkheadConfig(backendWithSharedConfig, compositeBulkheadCustomizer(),
211+
"backendWithSharedConfig");
207212
assertThat(bulkhead2).isNotNull();
208213
assertThat(bulkhead2.getMaxConcurrentCalls()).isEqualTo(2);
209214
assertThat(bulkhead2.getMaxWaitDuration().toMillis()).isEqualTo(300L);
210215

211216
// Unknown backend should get default config of Registry
212217
BulkheadConfig bulkhead3 = bulkheadConfigurationProperties
213-
.createBulkheadConfig(new BulkheadConfigurationProperties.InstanceProperties());
218+
.createBulkheadConfig(new BulkheadConfigurationProperties.InstanceProperties(),
219+
compositeBulkheadCustomizer(), "unknown");
214220
assertThat(bulkhead3).isNotNull();
215221
assertThat(bulkhead3.getMaxWaitDuration().toMillis()).isEqualTo(0L);
216222

@@ -226,7 +232,8 @@ public void testCreateBulkHeadPropertiesWithUnknownConfig() {
226232

227233
//When
228234
assertThatThrownBy(
229-
() -> bulkheadConfigurationProperties.createBulkheadConfig(instanceProperties))
235+
() -> bulkheadConfigurationProperties.createBulkheadConfig(instanceProperties,
236+
compositeBulkheadCustomizer(), "unknownConfig"))
230237
.isInstanceOf(ConfigurationNotFoundException.class)
231238
.hasMessage("Configuration with name 'unknownConfig' does not exist");
232239
}
@@ -255,4 +262,12 @@ public void testThreadPoolBulkheadIllegalArgumentOnEventConsumerBufferSize() {
255262
defaultProperties.setEventConsumerBufferSize(-1);
256263
}
257264

265+
private CompositeCustomizer<BulkheadConfigCustomizer> compositeBulkheadCustomizer() {
266+
return new CompositeCustomizer<>(Collections.emptyList());
267+
}
268+
269+
private CompositeCustomizer<ThreadPoolBulkheadConfigCustomizer> compositeThreadPoolBulkheadCustomizer() {
270+
return new CompositeCustomizer<>(Collections.emptyList());
271+
}
272+
258273
}

0 commit comments

Comments
 (0)