[
https://issues.apache.org/jira/browse/GOBBLIN-1061?focusedWorklogId=501151&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-501151
]
ASF GitHub Bot logged work on GOBBLIN-1061:
-------------------------------------------
Author: ASF GitHub Bot
Created on: 15/Oct/20 16:35
Start Date: 15/Oct/20 16:35
Worklog Time Spent: 10m
Work Description: sv2000 commented on a change in pull request #3124:
URL: https://github.com/apache/incubator-gobblin/pull/3124#discussion_r505677256
##########
File path:
gobblin-service/src/test/java/org/apache/gobblin/service/modules/core/GobblinServiceRedirectTest.java
##########
@@ -0,0 +1,301 @@
+/*
+ * 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.io.File;
+import java.net.InetAddress;
+import java.util.Map;
+import java.util.Properties;
+import java.util.UUID;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.curator.test.TestingServer;
+import org.apache.hadoop.fs.Path;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import com.google.common.base.Optional;
+import com.google.common.collect.Maps;
+import com.linkedin.data.template.StringMap;
+import com.linkedin.r2.transport.http.client.HttpClientFactory;
+import com.linkedin.restli.client.RestLiResponseException;
+
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.metastore.MysqlJobStatusStateStoreFactory;
+import org.apache.gobblin.metastore.testing.ITestMetastoreDatabase;
+import org.apache.gobblin.metastore.testing.TestMetastoreDatabaseFactory;
+import org.apache.gobblin.runtime.spec_catalog.FlowCatalog;
+import org.apache.gobblin.service.FlowConfig;
+import org.apache.gobblin.service.FlowConfigClient;
+import org.apache.gobblin.service.FlowId;
+import org.apache.gobblin.service.Schedule;
+import org.apache.gobblin.service.ServiceConfigKeys;
+import org.apache.gobblin.service.modules.utils.HelixUtils;
+import org.apache.gobblin.service.monitoring.FsJobStatusRetriever;
+import org.apache.gobblin.util.ConfigUtils;
+
+
+@Test
+public class GobblinServiceRedirectTest {
+
+ private static final Logger logger =
LoggerFactory.getLogger(GobblinServiceRedirectTest.class);
+
+ private static final String QUARTZ_INSTANCE_NAME =
"org.quartz.scheduler.instanceName";
+ private static final String QUARTZ_THREAD_POOL_COUNT =
"org.quartz.threadPool.threadCount";
+
+ private static final String COMMON_SPEC_STORE_PARENT_DIR =
"/tmp/serviceCoreCommon/";
+
+ private static final String NODE_1_SERVICE_WORK_DIR =
"/tmp/serviceWorkDirNode1/";
+ private static final String NODE_1_SPEC_STORE_PARENT_DIR =
"/tmp/serviceCoreNode1/";
+ private static final String NODE_1_TOPOLOGY_SPEC_STORE_DIR =
"/tmp/serviceCoreNode1/topologyTestSpecStoreNode1";
+ private static final String NODE_1_FLOW_SPEC_STORE_DIR =
"/tmp/serviceCoreCommon/flowTestSpecStore";
+ private static final String NODE_1_JOB_STATUS_STATE_STORE_DIR =
"/tmp/serviceCoreNode1/fsJobStatusRetriever";
+
+ private static final String NODE_2_SERVICE_WORK_DIR =
"/tmp/serviceWorkDirNode2/";
+ private static final String NODE_2_SPEC_STORE_PARENT_DIR =
"/tmp/serviceCoreNode2/";
+ private static final String NODE_2_TOPOLOGY_SPEC_STORE_DIR =
"/tmp/serviceCoreNode2/topologyTestSpecStoreNode2";
+ private static final String NODE_2_FLOW_SPEC_STORE_DIR =
"/tmp/serviceCoreCommon/flowTestSpecStore";
+ private static final String NODE_2_JOB_STATUS_STATE_STORE_DIR =
"/tmp/serviceCoreNode2/fsJobStatusRetriever";
+
+ private static final String TEST_HELIX_CLUSTER_NAME =
"testGobblinServiceCluster";
+
+ private static final String TEST_GROUP_NAME_1 = "testGroup1";
+ private static final String TEST_FLOW_NAME_1 = "testFlow1";
+ private static final String TEST_SCHEDULE_1 = "0 1/0 * ? * *";
+ private static final String TEST_TEMPLATE_URI_1 =
"FS:///templates/test.template";
+
+ private static final String TEST_GROUP_NAME_2 = "testGroup2";
+ private static final String TEST_FLOW_NAME_2 = "testFlow2";
+ private static final String TEST_SCHEDULE_2 = "0 1/0 * ? * *";
+ private static final String TEST_TEMPLATE_URI_2 =
"FS:///templates/test.template";
+
+ private static final String TEST_GOBBLIN_EXECUTOR_NAME =
"testGobblinExecutor";
+ private static final String TEST_SOURCE_NAME = "testSource";
+ private static final String TEST_SINK_NAME = "testSink";
+
+ private static final String PORT1 = "1000";
+ private static final String PORT2 = "2000";
+ private static final String PREFIX = "https://";;
+ private static final String SERVICE_NAME = "gobblinServiceTest";
+
+ private GobblinServiceManager node1GobblinServiceManager;
+ private FlowConfigClient node1FlowConfigClient;
+
+ private GobblinServiceManager node2GobblinServiceManager;
+ private FlowConfigClient node2FlowConfigClient;
+
+ private TestingServer testingZKServer;
+
+ private Properties node1ServiceCoreProperties;
+ private Properties node2ServiceCoreProperties;
+
+ @BeforeClass
+ public void setup() throws Exception {
+ // Clean up common Flow Spec Dir
+ cleanUpDir(COMMON_SPEC_STORE_PARENT_DIR);
+
+ // Clean up work dir for Node 1
+ cleanUpDir(NODE_1_SERVICE_WORK_DIR);
+ cleanUpDir(NODE_1_SPEC_STORE_PARENT_DIR);
+
+ // Clean up work dir for Node 2
+ cleanUpDir(NODE_2_SERVICE_WORK_DIR);
+ cleanUpDir(NODE_2_SPEC_STORE_PARENT_DIR);
+
+ // Use a random ZK port
+ this.testingZKServer = new TestingServer(-1);
+ logger.info("Testing ZK Server listening on: " +
testingZKServer.getConnectString());
+ HelixUtils.createGobblinHelixCluster(testingZKServer.getConnectString(),
TEST_HELIX_CLUSTER_NAME);
+
+ ITestMetastoreDatabase testMetastoreDatabase =
TestMetastoreDatabaseFactory.get();
+
+ Properties commonServiceCoreProperties = new Properties();
+
commonServiceCoreProperties.put(ServiceConfigKeys.ZK_CONNECTION_STRING_KEY,
testingZKServer.getConnectString());
+ commonServiceCoreProperties.put(ServiceConfigKeys.HELIX_CLUSTER_NAME_KEY,
TEST_HELIX_CLUSTER_NAME);
+ commonServiceCoreProperties.put(ServiceConfigKeys.HELIX_INSTANCE_NAME_KEY,
"GaaS_" + UUID.randomUUID().toString());
+
commonServiceCoreProperties.put(ServiceConfigKeys.TOPOLOGY_FACTORY_TOPOLOGY_NAMES_KEY
, TEST_GOBBLIN_EXECUTOR_NAME);
+ commonServiceCoreProperties.put(ServiceConfigKeys.TOPOLOGY_FACTORY_PREFIX
+ TEST_GOBBLIN_EXECUTOR_NAME + ".description",
+ "StandaloneTestExecutor");
+ commonServiceCoreProperties.put(ServiceConfigKeys.TOPOLOGY_FACTORY_PREFIX
+ TEST_GOBBLIN_EXECUTOR_NAME + ".version",
+ "1");
+ commonServiceCoreProperties.put(ServiceConfigKeys.TOPOLOGY_FACTORY_PREFIX
+ TEST_GOBBLIN_EXECUTOR_NAME + ".uri",
+ "gobblinExecutor");
+ commonServiceCoreProperties.put(ServiceConfigKeys.TOPOLOGY_FACTORY_PREFIX
+ TEST_GOBBLIN_EXECUTOR_NAME + ".specExecutorInstance",
+ "org.gobblin.service.InMemorySpecExecutor");
+ commonServiceCoreProperties.put(ServiceConfigKeys.TOPOLOGY_FACTORY_PREFIX
+ TEST_GOBBLIN_EXECUTOR_NAME + ".specExecInstance.capabilities",
+ TEST_SOURCE_NAME + ":" + TEST_SINK_NAME);
+ commonServiceCoreProperties.put(ConfigurationKeys.STATE_STORE_DB_USER_KEY,
"testUser");
+
commonServiceCoreProperties.put(ConfigurationKeys.STATE_STORE_DB_PASSWORD_KEY,
"testPassword");
+ commonServiceCoreProperties.put(ConfigurationKeys.STATE_STORE_DB_URL_KEY,
testMetastoreDatabase.getJdbcUrl());
+ commonServiceCoreProperties.put("zookeeper.connect",
testingZKServer.getConnectString());
+
commonServiceCoreProperties.put(ConfigurationKeys.STATE_STORE_FACTORY_CLASS_KEY,
MysqlJobStatusStateStoreFactory.class.getName());
+
commonServiceCoreProperties.put(ServiceConfigKeys.GOBBLIN_SERVICE_JOB_STATUS_MONITOR_ENABLED_KEY,
false);
+
+ commonServiceCoreProperties.put(ServiceConfigKeys.FORCE_LEADER, true);
+ commonServiceCoreProperties.put(ServiceConfigKeys.SERVICE_URL_PREFIX,
PREFIX);
+ commonServiceCoreProperties.put(ServiceConfigKeys.SERVICE_NAME,
SERVICE_NAME);
+
+ node1ServiceCoreProperties = new Properties();
+ node1ServiceCoreProperties.putAll(commonServiceCoreProperties);
+
node1ServiceCoreProperties.put(ConfigurationKeys.TOPOLOGYSPEC_STORE_DIR_KEY,
NODE_1_TOPOLOGY_SPEC_STORE_DIR);
+ node1ServiceCoreProperties.put(FlowCatalog.FLOWSPEC_STORE_DIR_KEY,
NODE_1_FLOW_SPEC_STORE_DIR);
+ node1ServiceCoreProperties.put(FsJobStatusRetriever.CONF_PREFIX + "." +
ConfigurationKeys.STATE_STORE_ROOT_DIR_KEY, NODE_1_JOB_STATUS_STATE_STORE_DIR);
+ node1ServiceCoreProperties.put(QUARTZ_INSTANCE_NAME, "QuartzScheduler1");
+ node1ServiceCoreProperties.put(QUARTZ_THREAD_POOL_COUNT, 3);
+ node1ServiceCoreProperties.put(ServiceConfigKeys.SERVICE_PORT, PORT1);
+
+ node2ServiceCoreProperties = new Properties();
+ node2ServiceCoreProperties.putAll(commonServiceCoreProperties);
+
node2ServiceCoreProperties.put(ConfigurationKeys.TOPOLOGYSPEC_STORE_DIR_KEY,
NODE_2_TOPOLOGY_SPEC_STORE_DIR);
+ node2ServiceCoreProperties.put(FlowCatalog.FLOWSPEC_STORE_DIR_KEY,
NODE_2_FLOW_SPEC_STORE_DIR);
+ node2ServiceCoreProperties.put(FsJobStatusRetriever.CONF_PREFIX + "." +
ConfigurationKeys.STATE_STORE_ROOT_DIR_KEY, NODE_2_JOB_STATUS_STATE_STORE_DIR);
+ node2ServiceCoreProperties.put(QUARTZ_INSTANCE_NAME, "QuartzScheduler2");
+ node2ServiceCoreProperties.put(QUARTZ_THREAD_POOL_COUNT, 3);
+ node2ServiceCoreProperties.put(ServiceConfigKeys.SERVICE_PORT, PORT2);
Review comment:
Same comment as above.
##########
File path:
gobblin-service/src/test/java/org/apache/gobblin/service/modules/core/GobblinServiceRedirectTest.java
##########
@@ -0,0 +1,301 @@
+/*
+ * 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.io.File;
+import java.net.InetAddress;
+import java.util.Map;
+import java.util.Properties;
+import java.util.UUID;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.curator.test.TestingServer;
+import org.apache.hadoop.fs.Path;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import com.google.common.base.Optional;
+import com.google.common.collect.Maps;
+import com.linkedin.data.template.StringMap;
+import com.linkedin.r2.transport.http.client.HttpClientFactory;
+import com.linkedin.restli.client.RestLiResponseException;
+
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.metastore.MysqlJobStatusStateStoreFactory;
+import org.apache.gobblin.metastore.testing.ITestMetastoreDatabase;
+import org.apache.gobblin.metastore.testing.TestMetastoreDatabaseFactory;
+import org.apache.gobblin.runtime.spec_catalog.FlowCatalog;
+import org.apache.gobblin.service.FlowConfig;
+import org.apache.gobblin.service.FlowConfigClient;
+import org.apache.gobblin.service.FlowId;
+import org.apache.gobblin.service.Schedule;
+import org.apache.gobblin.service.ServiceConfigKeys;
+import org.apache.gobblin.service.modules.utils.HelixUtils;
+import org.apache.gobblin.service.monitoring.FsJobStatusRetriever;
+import org.apache.gobblin.util.ConfigUtils;
+
+
+@Test
+public class GobblinServiceRedirectTest {
+
+ private static final Logger logger =
LoggerFactory.getLogger(GobblinServiceRedirectTest.class);
+
+ private static final String QUARTZ_INSTANCE_NAME =
"org.quartz.scheduler.instanceName";
+ private static final String QUARTZ_THREAD_POOL_COUNT =
"org.quartz.threadPool.threadCount";
+
+ private static final String COMMON_SPEC_STORE_PARENT_DIR =
"/tmp/serviceCoreCommon/";
+
+ private static final String NODE_1_SERVICE_WORK_DIR =
"/tmp/serviceWorkDirNode1/";
+ private static final String NODE_1_SPEC_STORE_PARENT_DIR =
"/tmp/serviceCoreNode1/";
+ private static final String NODE_1_TOPOLOGY_SPEC_STORE_DIR =
"/tmp/serviceCoreNode1/topologyTestSpecStoreNode1";
+ private static final String NODE_1_FLOW_SPEC_STORE_DIR =
"/tmp/serviceCoreCommon/flowTestSpecStore";
+ private static final String NODE_1_JOB_STATUS_STATE_STORE_DIR =
"/tmp/serviceCoreNode1/fsJobStatusRetriever";
+
+ private static final String NODE_2_SERVICE_WORK_DIR =
"/tmp/serviceWorkDirNode2/";
+ private static final String NODE_2_SPEC_STORE_PARENT_DIR =
"/tmp/serviceCoreNode2/";
+ private static final String NODE_2_TOPOLOGY_SPEC_STORE_DIR =
"/tmp/serviceCoreNode2/topologyTestSpecStoreNode2";
+ private static final String NODE_2_FLOW_SPEC_STORE_DIR =
"/tmp/serviceCoreCommon/flowTestSpecStore";
+ private static final String NODE_2_JOB_STATUS_STATE_STORE_DIR =
"/tmp/serviceCoreNode2/fsJobStatusRetriever";
+
+ private static final String TEST_HELIX_CLUSTER_NAME =
"testGobblinServiceCluster";
+
+ private static final String TEST_GROUP_NAME_1 = "testGroup1";
+ private static final String TEST_FLOW_NAME_1 = "testFlow1";
+ private static final String TEST_SCHEDULE_1 = "0 1/0 * ? * *";
+ private static final String TEST_TEMPLATE_URI_1 =
"FS:///templates/test.template";
+
+ private static final String TEST_GROUP_NAME_2 = "testGroup2";
+ private static final String TEST_FLOW_NAME_2 = "testFlow2";
+ private static final String TEST_SCHEDULE_2 = "0 1/0 * ? * *";
+ private static final String TEST_TEMPLATE_URI_2 =
"FS:///templates/test.template";
+
+ private static final String TEST_GOBBLIN_EXECUTOR_NAME =
"testGobblinExecutor";
+ private static final String TEST_SOURCE_NAME = "testSource";
+ private static final String TEST_SINK_NAME = "testSink";
+
+ private static final String PORT1 = "1000";
+ private static final String PORT2 = "2000";
+ private static final String PREFIX = "https://";;
+ private static final String SERVICE_NAME = "gobblinServiceTest";
+
+ private GobblinServiceManager node1GobblinServiceManager;
+ private FlowConfigClient node1FlowConfigClient;
+
+ private GobblinServiceManager node2GobblinServiceManager;
+ private FlowConfigClient node2FlowConfigClient;
+
+ private TestingServer testingZKServer;
+
+ private Properties node1ServiceCoreProperties;
+ private Properties node2ServiceCoreProperties;
+
+ @BeforeClass
+ public void setup() throws Exception {
+ // Clean up common Flow Spec Dir
+ cleanUpDir(COMMON_SPEC_STORE_PARENT_DIR);
+
+ // Clean up work dir for Node 1
+ cleanUpDir(NODE_1_SERVICE_WORK_DIR);
+ cleanUpDir(NODE_1_SPEC_STORE_PARENT_DIR);
+
+ // Clean up work dir for Node 2
+ cleanUpDir(NODE_2_SERVICE_WORK_DIR);
+ cleanUpDir(NODE_2_SPEC_STORE_PARENT_DIR);
+
+ // Use a random ZK port
+ this.testingZKServer = new TestingServer(-1);
+ logger.info("Testing ZK Server listening on: " +
testingZKServer.getConnectString());
+ HelixUtils.createGobblinHelixCluster(testingZKServer.getConnectString(),
TEST_HELIX_CLUSTER_NAME);
+
+ ITestMetastoreDatabase testMetastoreDatabase =
TestMetastoreDatabaseFactory.get();
+
+ Properties commonServiceCoreProperties = new Properties();
+
commonServiceCoreProperties.put(ServiceConfigKeys.ZK_CONNECTION_STRING_KEY,
testingZKServer.getConnectString());
+ commonServiceCoreProperties.put(ServiceConfigKeys.HELIX_CLUSTER_NAME_KEY,
TEST_HELIX_CLUSTER_NAME);
+ commonServiceCoreProperties.put(ServiceConfigKeys.HELIX_INSTANCE_NAME_KEY,
"GaaS_" + UUID.randomUUID().toString());
+
commonServiceCoreProperties.put(ServiceConfigKeys.TOPOLOGY_FACTORY_TOPOLOGY_NAMES_KEY
, TEST_GOBBLIN_EXECUTOR_NAME);
+ commonServiceCoreProperties.put(ServiceConfigKeys.TOPOLOGY_FACTORY_PREFIX
+ TEST_GOBBLIN_EXECUTOR_NAME + ".description",
+ "StandaloneTestExecutor");
+ commonServiceCoreProperties.put(ServiceConfigKeys.TOPOLOGY_FACTORY_PREFIX
+ TEST_GOBBLIN_EXECUTOR_NAME + ".version",
+ "1");
+ commonServiceCoreProperties.put(ServiceConfigKeys.TOPOLOGY_FACTORY_PREFIX
+ TEST_GOBBLIN_EXECUTOR_NAME + ".uri",
+ "gobblinExecutor");
+ commonServiceCoreProperties.put(ServiceConfigKeys.TOPOLOGY_FACTORY_PREFIX
+ TEST_GOBBLIN_EXECUTOR_NAME + ".specExecutorInstance",
+ "org.gobblin.service.InMemorySpecExecutor");
+ commonServiceCoreProperties.put(ServiceConfigKeys.TOPOLOGY_FACTORY_PREFIX
+ TEST_GOBBLIN_EXECUTOR_NAME + ".specExecInstance.capabilities",
+ TEST_SOURCE_NAME + ":" + TEST_SINK_NAME);
+ commonServiceCoreProperties.put(ConfigurationKeys.STATE_STORE_DB_USER_KEY,
"testUser");
+
commonServiceCoreProperties.put(ConfigurationKeys.STATE_STORE_DB_PASSWORD_KEY,
"testPassword");
+ commonServiceCoreProperties.put(ConfigurationKeys.STATE_STORE_DB_URL_KEY,
testMetastoreDatabase.getJdbcUrl());
+ commonServiceCoreProperties.put("zookeeper.connect",
testingZKServer.getConnectString());
+
commonServiceCoreProperties.put(ConfigurationKeys.STATE_STORE_FACTORY_CLASS_KEY,
MysqlJobStatusStateStoreFactory.class.getName());
+
commonServiceCoreProperties.put(ServiceConfigKeys.GOBBLIN_SERVICE_JOB_STATUS_MONITOR_ENABLED_KEY,
false);
+
+ commonServiceCoreProperties.put(ServiceConfigKeys.FORCE_LEADER, true);
+ commonServiceCoreProperties.put(ServiceConfigKeys.SERVICE_URL_PREFIX,
PREFIX);
+ commonServiceCoreProperties.put(ServiceConfigKeys.SERVICE_NAME,
SERVICE_NAME);
+
+ node1ServiceCoreProperties = new Properties();
+ node1ServiceCoreProperties.putAll(commonServiceCoreProperties);
+
node1ServiceCoreProperties.put(ConfigurationKeys.TOPOLOGYSPEC_STORE_DIR_KEY,
NODE_1_TOPOLOGY_SPEC_STORE_DIR);
+ node1ServiceCoreProperties.put(FlowCatalog.FLOWSPEC_STORE_DIR_KEY,
NODE_1_FLOW_SPEC_STORE_DIR);
+ node1ServiceCoreProperties.put(FsJobStatusRetriever.CONF_PREFIX + "." +
ConfigurationKeys.STATE_STORE_ROOT_DIR_KEY, NODE_1_JOB_STATUS_STATE_STORE_DIR);
+ node1ServiceCoreProperties.put(QUARTZ_INSTANCE_NAME, "QuartzScheduler1");
+ node1ServiceCoreProperties.put(QUARTZ_THREAD_POOL_COUNT, 3);
+ node1ServiceCoreProperties.put(ServiceConfigKeys.SERVICE_PORT, PORT1);
Review comment:
Should this be restliServer.getPort() instead of PORT1?
##########
File path:
gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-server/src/main/java/org/apache/gobblin/service/FlowExecutionResource.java
##########
@@ -52,10 +55,19 @@
public class FlowExecutionResource extends
ComplexKeyResourceTemplate<FlowStatusId, EmptyRecord, FlowExecution> {
Review comment:
One naive question: Can we create a FlowExecutionResourceHandler
interface with impls inside gobblin-service module and inject the handler here?
Essentially, replicating the pattern we used for FlowConfigs resource, and
making the code uniform.
##########
File path:
gobblin-restli/gobblin-flow-config-service/gobblin-flow-config-service-server/src/main/java/org/apache/gobblin/service/HelixLeaderUtils.java
##########
@@ -0,0 +1,105 @@
+/*
+ * 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;
+
+import java.io.UnsupportedEncodingException;
+import java.net.InetAddress;
+import java.net.URLDecoder;
+import java.net.URLEncoder;
+import java.net.UnknownHostException;
+
+import org.apache.helix.HelixManager;
+import org.apache.helix.PropertyKey;
+import org.apache.helix.model.LiveInstance;
+
+import com.google.common.base.Optional;
+import com.google.common.collect.ImmutableMap;
+import com.linkedin.data.DataMap;
+import com.linkedin.restli.common.HttpStatus;
+import com.linkedin.restli.server.RestLiServiceException;
+import com.typesafe.config.Config;
+
+import lombok.extern.slf4j.Slf4j;
+
+import org.apache.gobblin.util.ConfigUtils;
+
+
+/**
+ * Utils for storing/parsing helix URL in the helix instance name
+ */
+@Slf4j
+public class HelixLeaderUtils {
+ public static String HELIX_INSTANCE_NAME_SEPARATOR = "@";
+
+ /**
Review comment:
Empty Javadoc here. You may want to remove it if not needed.
##########
File path:
gobblin-service/src/test/java/org/apache/gobblin/service/modules/core/GobblinServiceRedirectTest.java
##########
@@ -0,0 +1,301 @@
+/*
+ * 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.io.File;
+import java.net.InetAddress;
+import java.util.Map;
+import java.util.Properties;
+import java.util.UUID;
+
+import org.apache.commons.io.FileUtils;
+import org.apache.curator.test.TestingServer;
+import org.apache.hadoop.fs.Path;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import com.google.common.base.Optional;
+import com.google.common.collect.Maps;
+import com.linkedin.data.template.StringMap;
+import com.linkedin.r2.transport.http.client.HttpClientFactory;
+import com.linkedin.restli.client.RestLiResponseException;
+
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.metastore.MysqlJobStatusStateStoreFactory;
+import org.apache.gobblin.metastore.testing.ITestMetastoreDatabase;
+import org.apache.gobblin.metastore.testing.TestMetastoreDatabaseFactory;
+import org.apache.gobblin.runtime.spec_catalog.FlowCatalog;
+import org.apache.gobblin.service.FlowConfig;
+import org.apache.gobblin.service.FlowConfigClient;
+import org.apache.gobblin.service.FlowId;
+import org.apache.gobblin.service.Schedule;
+import org.apache.gobblin.service.ServiceConfigKeys;
+import org.apache.gobblin.service.modules.utils.HelixUtils;
+import org.apache.gobblin.service.monitoring.FsJobStatusRetriever;
+import org.apache.gobblin.util.ConfigUtils;
+
+
+@Test
+public class GobblinServiceRedirectTest {
+
+ private static final Logger logger =
LoggerFactory.getLogger(GobblinServiceRedirectTest.class);
+
+ private static final String QUARTZ_INSTANCE_NAME =
"org.quartz.scheduler.instanceName";
+ private static final String QUARTZ_THREAD_POOL_COUNT =
"org.quartz.threadPool.threadCount";
+
+ private static final String COMMON_SPEC_STORE_PARENT_DIR =
"/tmp/serviceCoreCommon/";
+
+ private static final String NODE_1_SERVICE_WORK_DIR =
"/tmp/serviceWorkDirNode1/";
Review comment:
Can we dynamically generate tmp dirs using Guava Files.createTempDir()
method? And then mark the file to be deleted using deleteOnExit(). Hardcoding
the location to /tmp has the potential to create flaky tests due to left over
state from previous builds in case of un-finished clean up. Also, it may end up
interfering with identical locations used in other test classes.
----------------------------------------------------------------
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]
Issue Time Tracking
-------------------
Worklog Id: (was: 501151)
Time Spent: 2h (was: 1h 50m)
> Kafka consumer Kerberos config
> ------------------------------
>
> Key: GOBBLIN-1061
> URL: https://issues.apache.org/jira/browse/GOBBLIN-1061
> Project: Apache Gobblin
> Issue Type: Bug
> Affects Versions: 0.15.0
> Environment: OS RHEL 7.5, Cloudera kafka 3.1.0 (kafka 1.0.1), CDH
> 5.16.2
> Reporter: Jan Tichý
> Priority: Blocker
> Attachments: application.conf, sip_voice_raw.pull
>
> Time Spent: 2h
> Remaining Estimate: 0h
>
> Does the gobblin support kerberos auth for kafka client consumer? I have
> right kafka consumer settings with sasl_plaintext configuration but if it
> tries to kinit it fails with exception below:
> {code:java}
> 2020-02-20 05:44:00 PST INFO [DefaultQuartzScheduler_Worker-1]
> org.apache.kafka.common.config.AbstractConfig - ConsumerConfig values:
> metric.reporters = []
> metadata.max.age.ms = 300000
> value.deserializer = class
> io.confluent.kafka.serializers.KafkaAvroDeserializer
> group.id = kafka09
> partition.assignment.strategy =
> [org.apache.kafka.clients.consumer.RangeAssignor]
> reconnect.backoff.ms = 50
> sasl.kerberos.ticket.renew.window.factor = 0.8
> max.partition.fetch.bytes = 1048576
> bootstrap.servers = [czrtim1hr.oskarmobil.cz:9092,
> czrtim2hr.oskarmobil.cz:9092]
> retry.backoff.ms = 100
> sasl.kerberos.kinit.cmd = /usr/bin/kinit
> sasl.kerberos.service.name = kafka
> sasl.kerberos.ticket.renew.jitter = 0.05
> ssl.keystore.type = JKS
> ssl.trustmanager.algorithm = PKIX
> enable.auto.commit = false
> ssl.key.password = null
> fetch.max.wait.ms = 500
> sasl.kerberos.min.time.before.relogin = 60000
> connections.max.idle.ms = 540000
> ssl.truststore.password = null
> session.timeout.ms = 30000
> metrics.num.samples = 2
> client.id =
> ssl.endpoint.identification.algorithm = null
> key.deserializer = class io.confluent.kafka.serializers.KafkaAvroDeserializer
> ssl.protocol = TLS
> check.crcs = true
> request.timeout.ms = 40000
> ssl.provider = null
> ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
> ssl.keystore.location = null
> heartbeat.interval.ms = 3000
> auto.commit.interval.ms = 5000
> receive.buffer.bytes = 32768
> ssl.cipher.suites = null
> ssl.truststore.type = JKS
> security.protocol = SASL_PLAINTEXT
> ssl.truststore.location = null
> ssl.keystore.password = null
> ssl.keymanager.algorithm = SunX509
> metrics.sample.window.ms = 30000
> fetch.min.bytes = 1
> send.buffer.bytes = 131072
> auto.offset.reset = latest2020-02-20 05:44:00 PST ERROR
> [DefaultQuartzScheduler_Worker-1] org.apache.gobblin.runtime.SourceDecorator
> - Failed to get work units for job job_SipVoiceRaw_1582206240042
> org.apache.kafka.common.KafkaException: Failed to construct kafka consumer
> at
> org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:648)
> at
> org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:542)
> at
> org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer.java:524)
> at
> org.apache.gobblin.kafka.client.Kafka09ConsumerClient.<init>(Kafka09ConsumerClient.java:116)
> at
> org.apache.gobblin.kafka.client.Kafka09ConsumerClient.<init>(Kafka09ConsumerClient.java:69)
> at
> org.apache.gobblin.kafka.client.Kafka09ConsumerClient$Factory.create(Kafka09ConsumerClient.java:224)
> at
> org.apache.gobblin.source.extractor.extract.kafka.KafkaSource.getWorkunits(KafkaSource.java:210)
> at
> org.apache.gobblin.runtime.SourceDecorator.getWorkunitStream(SourceDecorator.java:81)
> at
> org.apache.gobblin.runtime.AbstractJobLauncher.launchJob(AbstractJobLauncher.java:411)
> at
> org.apache.gobblin.cluster.GobblinHelixJobLauncher.launchJob(GobblinHelixJobLauncher.java:378)
> at org.apache.gobblin.scheduler.JobScheduler.runJob(JobScheduler.java:487)
> at
> org.apache.gobblin.cluster.HelixRetriggeringJobCallable.runJobLauncherLoop(HelixRetriggeringJobCallable.java:203)
> at
> org.apache.gobblin.cluster.HelixRetriggeringJobCallable.call(HelixRetriggeringJobCallable.java:159)
> at
> org.apache.gobblin.cluster.GobblinHelixJobScheduler.runJob(GobblinHelixJobScheduler.java:228)
> at
> org.apache.gobblin.cluster.GobblinHelixJob.executeImpl(GobblinHelixJob.java:61)
> at org.apache.gobblin.scheduler.BaseGobblinJob.execute(BaseGobblinJob.java:58)
> at org.quartz.core.JobRunShell.run(JobRunShell.java:202)
> at
> org.quartz.simpl.SimpleThreadPool$WorkerThread.run(SimpleThreadPool.java:573)
> Caused by: org.apache.kafka.common.KafkaException:
> java.lang.IllegalArgumentException: You must pass
> java.security.auth.login.config in secure mode.
> at
> org.apache.kafka.common.network.SaslChannelBuilder.configure(SaslChannelBuilder.java:74)
> at
> org.apache.kafka.common.network.ChannelBuilders.create(ChannelBuilders.java:60){code}
> I have these conf options in my config for kerberos auth:
> gobblin.yarn.keytab.file.path="/home/morpheus2/.keytab"
> gobblin.yarn.keytab.principal.name=morpheus2
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
