autumnust commented on a change in pull request #3281:
URL: https://github.com/apache/gobblin/pull/3281#discussion_r646779909
##########
File path:
gobblin-service/src/main/java/org/apache/gobblin/service/modules/core/GobblinServiceManager.java
##########
@@ -673,9 +575,15 @@ public static void main(String[] args) throws Exception {
}
Config config = ConfigFactory.load();
- try (GobblinServiceManager gobblinServiceManager = new
GobblinServiceManager(
- cmd.getOptionValue(SERVICE_NAME_OPTION_NAME), getServiceId(cmd),
- config, Optional.<Path>absent())) {
+
+ GobblinServiceConfiguration serviceConfiguration =
+ new
GobblinServiceConfiguration(cmd.getOptionValue(SERVICE_NAME_OPTION_NAME),
getServiceId(cmd), config,
+ null);
+
+ GobblinServiceGuiceModule guiceModule = new
GobblinServiceGuiceModule(serviceConfiguration);
+ Injector injector = Guice.createInjector(guiceModule);
+
+ try (GobblinServiceManager gobblinServiceManager =
injector.getInstance(GobblinServiceManager.class)) {
Review comment:
Much cleaner process to initialize serviceManager !
##########
File path:
gobblin-service/src/main/java/org/apache/gobblin/service/modules/core/GobblinServiceConfiguration.java
##########
@@ -0,0 +1,155 @@
+/*
+ * 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.gobblin.service.modules.core;
+
+import java.util.Objects;
+
+import org.apache.hadoop.fs.Path;
+
+import com.typesafe.config.Config;
+
+import javax.annotation.Nullable;
+
+import org.apache.gobblin.service.ServiceConfigKeys;
+import org.apache.gobblin.util.ConfigUtils;
+
+
+public class GobblinServiceConfiguration {
+ private final String serviceName;
+ private final String serviceId;
+
+ private final boolean isTopologyCatalogEnabled;
+ private final boolean isFlowCatalogEnabled;
+ private final boolean isSchedulerEnabled;
+ private final boolean isRestLIServerEnabled;
+ private final boolean isTopologySpecFactoryEnabled;
+ private final boolean isGitConfigMonitorEnabled;
+ private final boolean isDagManagerEnabled;
+ private final boolean isJobStatusMonitorEnabled;
+ private final boolean isHelixManagerEnabled;
+ private final boolean flowCatalogLocalCommit;
+
+ private final Config config;
+
+ @Nullable
+ private final Path serviceWorkDir;
+
+ public GobblinServiceConfiguration(String serviceName, String serviceId,
Config config,
+ @Nullable Path serviceWorkDir) {
+ this.serviceName = Objects.requireNonNull(serviceName,"Service name cannot
be null");
+ this.serviceId = Objects.requireNonNull(serviceId,"Service id cannot be
null");
+ this.config = Objects.requireNonNull(config, "Config cannot be null");
+ this.serviceWorkDir = serviceWorkDir;
+
+ isTopologyCatalogEnabled =
+ ConfigUtils.getBoolean(config,
ServiceConfigKeys.GOBBLIN_SERVICE_TOPOLOGY_CATALOG_ENABLED_KEY, true);
+ isFlowCatalogEnabled =
+ ConfigUtils.getBoolean(config,
ServiceConfigKeys.GOBBLIN_SERVICE_FLOW_CATALOG_ENABLED_KEY, true);
+
+ if (isFlowCatalogEnabled) {
+ flowCatalogLocalCommit =
+ ConfigUtils.getBoolean(config,
ServiceConfigKeys.GOBBLIN_SERVICE_FLOW_CATALOG_LOCAL_COMMIT,
+
ServiceConfigKeys.DEFAULT_GOBBLIN_SERVICE_FLOW_CATALOG_LOCAL_COMMIT);
+ isGitConfigMonitorEnabled =
+ ConfigUtils.getBoolean(config,
ServiceConfigKeys.GOBBLIN_SERVICE_GIT_CONFIG_MONITOR_ENABLED_KEY, false);
+ } else {
+ flowCatalogLocalCommit = false;
+ isGitConfigMonitorEnabled = false;
+ }
+
+ this.isHelixManagerEnabled =
config.hasPath(ServiceConfigKeys.ZK_CONNECTION_STRING_KEY);
+ this.isDagManagerEnabled =
+ ConfigUtils.getBoolean(config,
ServiceConfigKeys.GOBBLIN_SERVICE_DAG_MANAGER_ENABLED_KEY, false);
+ this.isJobStatusMonitorEnabled =
+ ConfigUtils.getBoolean(config,
ServiceConfigKeys.GOBBLIN_SERVICE_JOB_STATUS_MONITOR_ENABLED_KEY, true);
+ this.isSchedulerEnabled =
+ ConfigUtils.getBoolean(config,
ServiceConfigKeys.GOBBLIN_SERVICE_SCHEDULER_ENABLED_KEY, true);
+ this.isRestLIServerEnabled =
+ ConfigUtils.getBoolean(config,
ServiceConfigKeys.GOBBLIN_SERVICE_RESTLI_SERVER_ENABLED_KEY, true);
+ this.isTopologySpecFactoryEnabled =
+ ConfigUtils.getBoolean(config,
ServiceConfigKeys.GOBBLIN_SERVICE_TOPOLOGY_SPEC_FACTORY_ENABLED_KEY, true);
+ }
+
+ public String getServiceName() {
Review comment:
Is that possible to use lombok annotation to avoid these boilerplate
code ?
##########
File path:
gobblin-service/src/main/java/org/apache/gobblin/service/modules/core/GobblinServiceManager.java
##########
@@ -471,21 +310,83 @@ private void handleLeadershipChange(NotificationContext
changeContext) {
helixLeaderGauges.get().setState(LeaderState.SLAVE);
}
- if (this.isGitConfigMonitorEnabled) {
+ if (configuration.isGitConfigMonitorEnabled()) {
this.gitConfigMonitor.setActive(false);
}
- if (this.isDagManagerEnabled) {
+ if (configuration.isDagManagerEnabled()) {
this.dagManager.setActive(false);
this.eventBus.unregister(this.dagManager);
}
}
}
+ private void registerServicesInLauncher(){
+ if (configuration.isTopologyCatalogEnabled()) {
+ this.serviceLauncher.addService(topologyCatalog);
+ }
+
+ if (configuration.isFlowCatalogEnabled()) {
+ this.serviceLauncher.addService(flowCatalog);
+
+ if (configuration.isGitConfigMonitorEnabled()) {
+ this.serviceLauncher.addService(gitConfigMonitor);
+ }
+ }
+
+ if (configuration.isDagManagerEnabled()) {
+ this.serviceLauncher.addService(dagManager);
+ }
+
+ if (configuration.isJobStatusMonitorEnabled()) {
+ this.serviceLauncher.addService(jobStatusMonitor);
+ }
+
+ if (configuration.isSchedulerEnabled()) {
+ this.serviceLauncher.addService(schedulerService);
+ this.serviceLauncher.addService(scheduler);
+ }
+
+ if (configuration.isRestLIServerEnabled()) {
+ this.serviceLauncher.addService(restliServer);
+ }
+ }
+
+ private void configureServices(){
+ if (configuration.isRestLIServerEnabled()) {
+ this.restliServer = EmbeddedRestliServer.builder()
+ .resources(Lists.newArrayList(FlowConfigsResource.class,
FlowConfigsV2Resource.class))
+ .injector(injector)
+ .build();
+
+ if
(configuration.getInnerConfig().hasPath(ServiceConfigKeys.SERVICE_PORT)) {
+
this.restliServer.setPort(configuration.getInnerConfig().getInt(ServiceConfigKeys.SERVICE_PORT));
+ }
+ }
+
+ registerServicesInLauncher();
+
+ // Register Scheduler to listen to changes in Flows
+ if (configuration.isSchedulerEnabled()) {
+ this.flowCatalog.addListener(this.scheduler);
+ }
+ }
+
+ void ensureInjected() {
Review comment:
private?
##########
File path:
gobblin-service/src/main/java/org/apache/gobblin/service/modules/core/ControllerUserDefinedMessageHandlerFactory.java
##########
@@ -51,7 +51,7 @@
class ControllerUserDefinedMessageHandlerFactory implements
MessageHandlerFactory {
private boolean flowCatalogLocalCommit;
private GobblinServiceJobScheduler jobScheduler;
- private GobblinServiceFlowConfigResourceHandler resourceHandler;
+ private FlowConfigsResourceHandler resourceHandler;
Review comment:
Just to my own understanding: What's the indication for this type change
? (Am asking since this change doesn't seem to be related to the major theme
of this PR)
##########
File path:
gobblin-service/src/main/java/org/apache/gobblin/service/modules/utils/InjectionNames.java
##########
@@ -0,0 +1,24 @@
+/*
+ * 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.gobblin.service.modules.utils;
+
+public final class InjectionNames {
Review comment:
What's the reason for this additional class?
##########
File path:
gobblin-service/src/main/java/org/apache/gobblin/service/modules/orchestration/DagManager.java
##########
@@ -192,7 +193,7 @@ public String toString() {
private volatile boolean isActive = false;
- public DagManager(Config config, boolean instrumentationEnabled) {
+ public DagManager(Config config, JobStatusRetriever jobStatusRetriever,
boolean instrumentationEnabled) {
Review comment:
Is backward compatibility a concern here?
--
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.
For queries about this service, please contact Infrastructure at:
[email protected]
