forked from apache/samza
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
SAMZA-2790: Cleanup RunLoop constructor explosion (apache#1680)
Description: Runloop currently takes in lot of parameters and the constructor has grown to the point where it is unmanageable with multiple overloads. Introducing new configuration requires lot of updates to existing tests and components even if the parameters have no effect on all of the usages. With this PR, we should be able to decouple different users of RunLoop and enable these components to have their own scoped config. e.g., SideInputManager can now have its own set of runloop parameters without having to tie itself with TaskConfig. Changes: Introduce RunLoopConfig, a container object to hold all required parameters for runloop from Config. Remove existing overloads of constructor Simplify the constructor to take RunLoopConfig and initialize the necessary components and fields Introduce SideInputManagerRunLoopConfig, an overload of RunLoopConfig to be used within SideInputManager Modify RunLoopFactory create method signature Clean up ApplicationUtil and moved the method to ApplicationConfig and added unit tests API Changes: No external API change
- Loading branch information
1 parent
fa4008e
commit aeeaf0b
Showing
10 changed files
with
293 additions
and
191 deletions.
There are no files selected for viewing
This file contains 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
This file contains 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
84 changes: 84 additions & 0 deletions
84
samza-core/src/main/java/org/apache/samza/config/RunLoopConfig.java
This file contains 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,84 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one | ||
| * or more contributor license agreements. See the NOTICE file | ||
| * distributed with this work for additional information | ||
| * regarding copyright ownership. The ASF licenses this file | ||
| * to you under the Apache License, Version 2.0 (the | ||
| * "License"); you may not use this file except in compliance | ||
| * with the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, | ||
| * software distributed under the License is distributed on an | ||
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | ||
| * KIND, either express or implied. See the License for the | ||
| * specific language governing permissions and limitations | ||
| * under the License. | ||
| */ | ||
| package org.apache.samza.config; | ||
|
|
||
| import java.util.concurrent.TimeUnit; | ||
|
|
||
|
|
||
| /** | ||
| * A container class to hold run loop related configurations to prevent constructor explosion | ||
| * in {@link org.apache.samza.container.RunLoop} | ||
| */ | ||
| public class RunLoopConfig extends MapConfig { | ||
| private static final String CONTAINER_DISK_QUOTA_DELAY_MAX_MS = "container.disk.quota.delay.max.ms"; | ||
| private ApplicationConfig appConfig; | ||
| private JobConfig jobConfig; | ||
| private TaskConfig taskConfig; | ||
|
|
||
| public RunLoopConfig(Config config) { | ||
| super(config); | ||
| this.appConfig = new ApplicationConfig(config); | ||
| this.jobConfig = new JobConfig(config); | ||
| this.taskConfig = new TaskConfig(config); | ||
| } | ||
|
|
||
| public int getMaxConcurrency() { | ||
| return taskConfig.getMaxConcurrency(); | ||
| } | ||
|
|
||
| public long getTaskCallbackTimeoutMs() { | ||
| return taskConfig.getCallbackTimeoutMs(); | ||
| } | ||
|
|
||
| public long getDrainCallbackTimeoutMs() { | ||
| return taskConfig.getDrainCallbackTimeoutMs(); | ||
| } | ||
|
|
||
| public boolean asyncCommitEnabled() { | ||
| return taskConfig.getAsyncCommit(); | ||
| } | ||
|
|
||
| public long getWindowMs() { | ||
| return taskConfig.getWindowMs(); | ||
| } | ||
|
|
||
| public long getCommitMs() { | ||
| return taskConfig.getCommitMs(); | ||
| } | ||
|
|
||
| public long getMaxIdleMs() { | ||
| return taskConfig.getMaxIdleMs(); | ||
| } | ||
|
|
||
| public long getMaxThrottlingDelayMs() { | ||
| return getLong(CONTAINER_DISK_QUOTA_DELAY_MAX_MS, TimeUnit.SECONDS.toMillis(1)); | ||
| } | ||
|
|
||
| public String getRunId() { | ||
| return appConfig.getRunId(); | ||
| } | ||
|
|
||
| public int getElasticityFactor() { | ||
| return jobConfig.getElasticityFactor(); | ||
| } | ||
|
|
||
| public boolean isHighLevelApiJob() { | ||
| return appConfig.isHighLevelApiJob(); | ||
| } | ||
| } |
This file contains 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
This file contains 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
This file contains 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
Oops, something went wrong.