MelodyShen commented on code in PR #30439:
URL: https://github.com/apache/beam/pull/30439#discussion_r1508318117
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/util/BoundedQueueExecutor.java:
##########
@@ -48,6 +49,7 @@ public BoundedQueueExecutor(
int maximumElementsOutstanding,
long maximumBytesOutstanding,
ThreadFactory threadFactory) {
+ this.maximumThreadCount.set(maximumPoolSize);
Review Comment:
Changed to int and use synchronized instead.
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/util/BoundedQueueExecutor.java:
##########
@@ -38,6 +38,7 @@ public class BoundedQueueExecutor {
private int elementsOutstanding = 0;
private long bytesOutstanding = 0;
private final AtomicInteger activeCount = new AtomicInteger();
+ private final AtomicInteger maximumThreadCount = new AtomicInteger();
Review Comment:
Sure.
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorkerTest.java:
##########
@@ -2916,6 +2916,69 @@ public void testActiveThreadMetric() throws Exception {
executor.shutdown();
}
+ @Test
+ public void testOverrideMaximumThreadCount() throws Exception {
+ int maxThreads = 5;
+ int threadExpirationSec = 60;
Review Comment:
Cool!
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/util/BoundedQueueExecutor.java:
##########
@@ -70,7 +72,7 @@ protected void beforeExecute(Thread t, Runnable r) {
protected void afterExecute(Runnable r, Throwable t) {
super.afterExecute(r, t);
synchronized (this) {
- if (activeCount.getAndDecrement() == maximumPoolSize) {
+ if (activeCount.getAndDecrement() == maximumThreadCount.get()) {
Review Comment:
Yeah that makes sense.
##########
runners/google-cloud-dataflow-java/worker/src/test/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorkerTest.java:
##########
@@ -2916,6 +2916,69 @@ public void testActiveThreadMetric() throws Exception {
executor.shutdown();
}
+ @Test
+ public void testOverrideMaximumThreadCount() throws Exception {
+ int maxThreads = 5;
+ int threadExpirationSec = 60;
+ CountDownLatch processStart1 = new CountDownLatch(2);
+ AtomicBoolean stop = new AtomicBoolean(false);
+ // setting up actual implementation of executor instead of mocking to keep
track of
+ // active thread count.
+ BoundedQueueExecutor executor =
Review Comment:
Sounds good. Will refactor it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
