You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by el...@apache.org on 2019/07/09 11:40:58 UTC
[hadoop] branch ozone-0.4.1 updated: HDDS-1717. MR Job fails as
OMFailoverProxyProvider has dependency hadoop-3.2
This is an automated email from the ASF dual-hosted git repository.
elek pushed a commit to branch ozone-0.4.1
in repository https://gitbox.apache.org/repos/asf/hadoop.git
The following commit(s) were added to refs/heads/ozone-0.4.1 by this push:
new 0b2dd3f HDDS-1717. MR Job fails as OMFailoverProxyProvider has dependency hadoop-3.2
0b2dd3f is described below
commit 0b2dd3f7713c5c2d4719ad587b759caa998d3819
Author: Hanisha Koneru <ha...@apache.org>
AuthorDate: Tue Jul 9 13:22:00 2019 +0200
HDDS-1717. MR Job fails as OMFailoverProxyProvider has dependency hadoop-3.2
Closes #1056
(cherry picked from commit 8861573e8c97b0d040aeb9bdd71c6f0c62038af7)
---
.../apache/hadoop/ozone/client/rpc/RpcClient.java | 3 +-
.../ozone/om/ha/OMFailoverProxyProvider.java | 82 +++++++++-------------
.../org/apache/hadoop/ozone/om/ha/OMProxyInfo.java | 59 ++++++++++++++++
.../client/rpc/TestOzoneRpcClientAbstract.java | 4 +-
.../apache/hadoop/ozone/om/TestOzoneManagerHA.java | 10 +--
5 files changed, 100 insertions(+), 58 deletions(-)
diff --git a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java
index ffb8bce..ce7150d 100644
--- a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java
+++ b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java
@@ -223,8 +223,7 @@ public class RpcClient implements ClientProtocol, KeyProviderTokenIssuer {
retryInterval = OzoneUtils.getTimeDurationInMS(conf,
OzoneConfigKeys.OZONE_CLIENT_RETRY_INTERVAL,
OzoneConfigKeys.OZONE_CLIENT_RETRY_INTERVAL_DEFAULT);
- dtService =
- getOMProxyProvider().getProxy().getDelegationTokenService();
+ dtService = getOMProxyProvider().getCurrentProxyDelegationToken();
boolean isUnsafeByteOperationsEnabled = conf.getBoolean(
OzoneConfigKeys.OZONE_UNSAFEBYTEOPERATIONS_ENABLED,
OzoneConfigKeys.OZONE_UNSAFEBYTEOPERATIONS_ENABLED_DEFAULT);
diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/OMFailoverProxyProvider.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/OMFailoverProxyProvider.java
index 481f221..0e5c3c6 100644
--- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/OMFailoverProxyProvider.java
+++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/OMFailoverProxyProvider.java
@@ -30,7 +30,6 @@ import org.apache.hadoop.net.NetUtils;
import org.apache.hadoop.ozone.OmUtils;
import org.apache.hadoop.ozone.OzoneConsts;
import org.apache.hadoop.ozone.om.protocolPB.OzoneManagerProtocolPB;
-import org.apache.hadoop.security.SecurityUtil;
import org.apache.hadoop.security.UserGroupInformation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -59,7 +58,8 @@ public class OMFailoverProxyProvider implements
LoggerFactory.getLogger(OMFailoverProxyProvider.class);
// Map of OMNodeID to its proxy
- private Map<String, OMProxyInfo> omProxies;
+ private Map<String, ProxyInfo<OzoneManagerProtocolPB>> omProxies;
+ private Map<String, OMProxyInfo> omProxyInfos;
private List<String> omNodeIDList;
private String currentProxyOMNodeId;
@@ -80,33 +80,9 @@ public class OMFailoverProxyProvider implements
currentProxyOMNodeId = omNodeIDList.get(currentProxyIndex);
}
- /**
- * Class to store proxy information.
- */
- public class OMProxyInfo
- extends FailoverProxyProvider.ProxyInfo<OzoneManagerProtocolPB> {
- private InetSocketAddress address;
- private Text dtService;
-
- OMProxyInfo(OzoneManagerProtocolPB proxy, String proxyInfoStr,
- Text dtService,
- InetSocketAddress addr) {
- super(proxy, proxyInfoStr);
- this.address = addr;
- this.dtService = dtService;
- }
-
- public InetSocketAddress getAddress() {
- return address;
- }
-
- public Text getDelegationTokenService() {
- return dtService;
- }
- }
-
private void loadOMClientConfigs(Configuration config) throws IOException {
this.omProxies = new HashMap<>();
+ this.omProxyInfos = new HashMap<>();
this.omNodeIDList = new ArrayList<>();
Collection<String> omServiceIds = config.getTrimmedStringCollection(
@@ -130,25 +106,21 @@ public class OMFailoverProxyProvider implements
continue;
}
- InetSocketAddress addr = NetUtils.createSocketAddr(rpcAddrStr);
+ OMProxyInfo omProxyInfo = new OMProxyInfo(nodeId, rpcAddrStr);
+
+ if (omProxyInfo.getAddress() != null) {
- // Add the OM client proxy info to list of proxies
- if (addr != null) {
- Text dtService = SecurityUtil.buildTokenService(addr);
- StringBuilder proxyInfo = new StringBuilder()
- .append(nodeId).append("(")
- .append(NetUtils.getHostPortString(addr)).append(")");
- OMProxyInfo omProxyInfo = new OMProxyInfo(null,
- proxyInfo.toString(), dtService, addr);
+ ProxyInfo<OzoneManagerProtocolPB> proxyInfo =
+ new ProxyInfo(null, omProxyInfo.toString());
// For a non-HA OM setup, nodeId might be null. If so, we assign it
// a dummy value
if (nodeId == null) {
nodeId = OzoneConsts.OM_NODE_ID_DUMMY;
}
- omProxies.put(nodeId, omProxyInfo);
+ omProxies.put(nodeId, proxyInfo);
+ omProxyInfos.put(nodeId, omProxyInfo);
omNodeIDList.add(nodeId);
-
} else {
LOG.error("Failed to create OM proxy for {} at address {}",
nodeId, rpcAddrStr);
@@ -183,26 +155,31 @@ public class OMFailoverProxyProvider implements
* @return the OM proxy object to invoke methods upon
*/
@Override
- public synchronized OMProxyInfo getProxy() {
- OMProxyInfo currentOMProxyInfo = omProxies.get(currentProxyOMNodeId);
- createOMProxyIfNeeded(currentOMProxyInfo);
- return currentOMProxyInfo;
+ public synchronized ProxyInfo getProxy() {
+ ProxyInfo currentProxyInfo = omProxies.get(currentProxyOMNodeId);
+ createOMProxyIfNeeded(currentProxyInfo, currentProxyOMNodeId);
+ return currentProxyInfo;
}
/**
- * Creates OM proxy object if it does not already exist.
+ * Creates proxy object if it does not already exist.
*/
- private OMProxyInfo createOMProxyIfNeeded(OMProxyInfo proxyInfo) {
+ private void createOMProxyIfNeeded(ProxyInfo proxyInfo,
+ String nodeId) {
if (proxyInfo.proxy == null) {
+ InetSocketAddress address = omProxyInfos.get(nodeId).getAddress();
try {
- proxyInfo.proxy = createOMProxy(proxyInfo.address);
+ proxyInfo.proxy = createOMProxy(address);
} catch (IOException ioe) {
LOG.error("{} Failed to create RPC proxy to OM at {}",
- this.getClass().getSimpleName(), proxyInfo.address, ioe);
+ this.getClass().getSimpleName(), address, ioe);
throw new RuntimeException(ioe);
}
}
- return proxyInfo;
+ }
+
+ public synchronized Text getCurrentProxyDelegationToken() {
+ return omProxyInfos.get(currentProxyOMNodeId).getDelegationTokenService();
}
/**
@@ -269,7 +246,7 @@ public class OMFailoverProxyProvider implements
*/
@Override
public synchronized void close() throws IOException {
- for (OMProxyInfo proxy : omProxies.values()) {
+ for (ProxyInfo<OzoneManagerProtocolPB> proxy : omProxies.values()) {
OzoneManagerProtocolPB omProxy = proxy.proxy;
if (omProxy != null) {
RPC.stopProxy(omProxy);
@@ -278,8 +255,13 @@ public class OMFailoverProxyProvider implements
}
@VisibleForTesting
- public List<OMProxyInfo> getOMProxies() {
- return new ArrayList<>(omProxies.values());
+ public List<ProxyInfo> getOMProxies() {
+ return new ArrayList<ProxyInfo>(omProxies.values());
+ }
+
+ @VisibleForTesting
+ public List<OMProxyInfo> getOMProxyInfos() {
+ return new ArrayList<OMProxyInfo>(omProxyInfos.values());
}
}
diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/OMProxyInfo.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/OMProxyInfo.java
new file mode 100644
index 0000000..b429ca0
--- /dev/null
+++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/OMProxyInfo.java
@@ -0,0 +1,59 @@
+/**
+ * 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.hadoop.ozone.om.ha;
+
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.security.SecurityUtil;
+
+import java.net.InetSocketAddress;
+
+/**
+ * Class to store OM proxy information.
+ */
+public class OMProxyInfo {
+ private String nodeId;
+ private String rpcAddrStr;
+ private InetSocketAddress rpcAddr;
+ private Text dtService;
+
+ OMProxyInfo(String nodeID, String rpcAddress) {
+ this.nodeId = nodeID;
+ this.rpcAddrStr = rpcAddress;
+ this.rpcAddr = NetUtils.createSocketAddr(rpcAddrStr);
+ this.dtService = SecurityUtil.buildTokenService(rpcAddr);
+ }
+
+ public String toString() {
+ StringBuilder sb = new StringBuilder()
+ .append("nodeId=")
+ .append(nodeId)
+ .append(",nodeAddress=")
+ .append(rpcAddrStr);
+ return sb.toString();
+ }
+
+ public InetSocketAddress getAddress() {
+ return rpcAddr;
+ }
+
+ public Text getDelegationTokenService() {
+ return dtService;
+ }
+}
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneRpcClientAbstract.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneRpcClientAbstract.java
index d24b3da..868e04a 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneRpcClientAbstract.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneRpcClientAbstract.java
@@ -76,6 +76,7 @@ import org.apache.hadoop.ozone.om.OzoneManager;
import org.apache.hadoop.ozone.om.exceptions.OMException;
import org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes;
import org.apache.hadoop.ozone.om.ha.OMFailoverProxyProvider;
+import org.apache.hadoop.ozone.om.ha.OMProxyInfo;
import org.apache.hadoop.ozone.om.helpers.OmKeyArgs;
import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
@@ -205,8 +206,7 @@ public abstract class TestOzoneRpcClientAbstract {
public void testOMClientProxyProvider() {
OMFailoverProxyProvider omFailoverProxyProvider = store.getClientProxy()
.getOMProxyProvider();
- List<OMFailoverProxyProvider.OMProxyInfo> omProxies =
- omFailoverProxyProvider.getOMProxies();
+ List<OMProxyInfo> omProxies = omFailoverProxyProvider.getOMProxyInfos();
// For a non-HA OM service, there should be only one OM proxy.
Assert.assertEquals(1, omProxies.size());
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerHA.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerHA.java
index 71a384f..16eb700 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerHA.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerHA.java
@@ -34,8 +34,10 @@ import org.apache.hadoop.ozone.client.io.OzoneInputStream;
import org.apache.hadoop.ozone.client.io.OzoneOutputStream;
import org.apache.hadoop.ozone.om.exceptions.OMException;
import org.apache.hadoop.ozone.om.ha.OMFailoverProxyProvider;
+import org.apache.hadoop.ozone.om.ha.OMProxyInfo;
import org.apache.hadoop.ozone.om.helpers.OmMultipartInfo;
import org.apache.hadoop.ozone.om.helpers.OmMultipartUploadCompleteInfo;
+import org.apache.hadoop.ozone.om.protocolPB.OzoneManagerProtocolPB;
import org.apache.hadoop.test.GenericTestUtils;
import org.apache.hadoop.ozone.client.OzoneClientFactory;
import org.apache.hadoop.ozone.client.OzoneVolume;
@@ -463,8 +465,8 @@ public class TestOzoneManagerHA {
OzoneClient rpcClient = cluster.getRpcClient();
OMFailoverProxyProvider omFailoverProxyProvider =
rpcClient.getObjectStore().getClientProxy().getOMProxyProvider();
- List<OMFailoverProxyProvider.OMProxyInfo> omProxies =
- omFailoverProxyProvider.getOMProxies();
+ List<OMProxyInfo> omProxies =
+ omFailoverProxyProvider.getOMProxyInfos();
Assert.assertEquals(numOfOMs, omProxies.size());
@@ -472,7 +474,7 @@ public class TestOzoneManagerHA {
InetSocketAddress omRpcServerAddr =
cluster.getOzoneManager(i).getOmRpcServerAddr();
boolean omClientProxyExists = false;
- for (OMFailoverProxyProvider.OMProxyInfo omProxyInfo : omProxies) {
+ for (OMProxyInfo omProxyInfo : omProxies) {
if (omProxyInfo.getAddress().equals(omRpcServerAddr)) {
omClientProxyExists = true;
break;
@@ -533,7 +535,7 @@ public class TestOzoneManagerHA {
// Perform a manual failover of the proxy provider to move the
// currentProxyIndex to a node other than the leader OM.
omFailoverProxyProvider.performFailover(
- omFailoverProxyProvider.getProxy().proxy);
+ (OzoneManagerProtocolPB) omFailoverProxyProvider.getProxy().proxy);
String newProxyNodeId = omFailoverProxyProvider.getCurrentProxyOMNodeId();
Assert.assertNotEquals(leaderOMNodeId, newProxyNodeId);
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org