OOZIE-3542 amend Handle better old Hdfs implementations in ECPolicyDisabler (zsombor via kmarton)
diff --git a/release-log.txt b/release-log.txt
index 96f91d9..0f406ae 100644
--- a/release-log.txt
+++ b/release-log.txt
@@ -1,5 +1,6 @@
-- Oozie 5.2.0 release (trunk - unreleased)
+OOZIE-3542 amend Handle better old Hdfs implementations in ECPolicyDisabler (zsombor via kmarton)
OOZIE-3529 Oozie not supported for s3 as filesystem (dionusos via asalamon74)
OOZIE-3465 Migrate from commons-codec (matijhs via asalamon74)
OOZIE-3533 Flaky test TestXLogService.testLog4jReload (asalamon74 via kmarton)
diff --git a/tools/src/main/java/org/apache/oozie/tools/ECPolicyDisabler.java b/tools/src/main/java/org/apache/oozie/tools/ECPolicyDisabler.java
index 6d51342..e24ac7a 100644
--- a/tools/src/main/java/org/apache/oozie/tools/ECPolicyDisabler.java
+++ b/tools/src/main/java/org/apache/oozie/tools/ECPolicyDisabler.java
@@ -27,6 +27,8 @@
import org.apache.hadoop.ipc.RemoteException;
import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcResponseHeaderProto.RpcErrorCodeProto;
+import org.apache.commons.lang3.tuple.Pair;
+
/**
* Utility class which can disable Erasure Coding for a given path.
*
@@ -68,22 +70,23 @@
DistributedFileSystem dfs = (DistributedFileSystem) fs;
final Object replicationPolicy = getReplicationPolicy();
Method getErasureCodingPolicyMethod = getMethod(dfs, GETERASURECODINGPOLICY_METHOD);
- final Object currentECPolicy = invokeMethod(getErasureCodingPolicyMethod, dfs, path);
+ final Pair<Object,Result> currentECPolicy = safeInvokeMethod(getErasureCodingPolicyMethod, dfs, path);
+ if (currentECPolicy.getRight() != null) {
+ return currentECPolicy.getRight();
+ }
- if (currentECPolicy != replicationPolicy) {
+ if (currentECPolicy.getLeft() != replicationPolicy) {
Method setECPolicyMethod = getMethod(dfs, SETERASURECODINGPOLICY_METHOD);
Method policyGetNameMethod = getMethod(replicationPolicy, GETNAME_METHOD);
- String name = (String) invokeMethod(policyGetNameMethod, replicationPolicy);
+ Pair<String,Result> pairName = safeInvokeMethod(policyGetNameMethod, replicationPolicy);
+ if (pairName.getRight() != null) {
+ return pairName.getRight();
+ }
- try {
- invokeMethod(setECPolicyMethod, dfs, path, name);
- } catch (RuntimeException e) {
- RpcErrorCodeProto errorCode = unwrapRemote(e);
- if (errorCode == RpcErrorCodeProto.ERROR_NO_SUCH_METHOD) {
- return Result.NO_SUCH_METHOD;
- }
- throw e;
+ Pair<String,Result> result = safeInvokeMethod(setECPolicyMethod, dfs, path, pairName.getLeft());
+ if (result.getRight() != null) {
+ return result.getRight();
}
return Result.DONE;
} else {
@@ -94,6 +97,18 @@
}
}
+ private static <X> Pair<X, Result> safeInvokeMethod(Method method, Object instance, Object... args) {
+ try {
+ return Pair.of((X) invokeMethod(method, instance, args), null);
+ } catch (RuntimeException e) {
+ RpcErrorCodeProto errorCode = unwrapRemote(e);
+ if (errorCode == RpcErrorCodeProto.ERROR_NO_SUCH_METHOD) {
+ return Pair.of(null, Result.NO_SUCH_METHOD);
+ }
+ throw e;
+ }
+ }
+
private static RpcErrorCodeProto unwrapRemote(Throwable e) {
if (e instanceof InvocationTargetException) {
return unwrapRemote(e.getCause());
diff --git a/tools/src/test/java/org/apache/hadoop/hdfs/protocol/SystemErasureCodingPolicies.java b/tools/src/test/java/org/apache/hadoop/hdfs/protocol/SystemErasureCodingPolicies.java
index ca6b65a..d5c5f9e 100644
--- a/tools/src/test/java/org/apache/hadoop/hdfs/protocol/SystemErasureCodingPolicies.java
+++ b/tools/src/test/java/org/apache/hadoop/hdfs/protocol/SystemErasureCodingPolicies.java
@@ -24,16 +24,26 @@
*
*/
public class SystemErasureCodingPolicies {
- public enum ReplicationPolicy {
- DEFAULT, OTHER;
+ private static ReplicationPolicy systemPolicy = ReplicationPolicy.DEFAULT;
+
+ public static class ReplicationPolicy {
+ public final static ReplicationPolicy DEFAULT = new ReplicationPolicy("DEFAULT");
+ public final static ReplicationPolicy OTHER = new ReplicationPolicy("OTHER");
+ private String name;
+ public ReplicationPolicy(String name) {
+ this.name = name;
+ }
public String getName() {
- return name();
+ return name;
}
}
public static ReplicationPolicy getReplicationPolicy() {
- return ReplicationPolicy.DEFAULT;
+ return systemPolicy;
}
+ public static void setSystemPolicy(ReplicationPolicy systemPolicy) {
+ SystemErasureCodingPolicies.systemPolicy = systemPolicy;
+ }
}
diff --git a/tools/src/test/java/org/apache/oozie/tools/TestECPolicyDisabler.java b/tools/src/test/java/org/apache/oozie/tools/TestECPolicyDisabler.java
index b085eb2..c2ba314 100644
--- a/tools/src/test/java/org/apache/oozie/tools/TestECPolicyDisabler.java
+++ b/tools/src/test/java/org/apache/oozie/tools/TestECPolicyDisabler.java
@@ -36,6 +36,7 @@
import org.apache.hadoop.ipc.protobuf.RpcHeaderProtos.RpcResponseHeaderProto.RpcErrorCodeProto;
import org.apache.oozie.tools.ECPolicyDisabler.Result;
import org.junit.Assert;
+import org.junit.Before;
import org.junit.Test;
import org.mockito.Mockito;
@@ -50,6 +51,11 @@
public abstract void setErasureCodingPolicy(Path path, String policy);
}
+ @Before
+ public void setup() {
+ SystemErasureCodingPolicies.setSystemPolicy(ReplicationPolicy.DEFAULT);
+ }
+
@Test
public void testNotSupported() {
FileSystem fs = mock(FileSystem.class);
@@ -91,6 +97,42 @@
}
@Test
+ public void testServerNotSupportsGetErasureCodingPolicyMethod() {
+ MockDistributedFileSystem fs = mock(MockDistributedFileSystem.class);
+ when(fs.getErasureCodingPolicy(any(Path.class))).thenThrow(createNoSuchMethodException());
+ ECPolicyDisabler.Result result = ECPolicyDisabler.check(fs, mock(Path.class));
+ assertEquals("result is expected", Result.NO_SUCH_METHOD, result);
+ verify(fs).getErasureCodingPolicy(any(Path.class));
+ verifyNoMoreInteractions(fs);
+ }
+
+ @Test
+ public void testServerNotSupportsGetName() {
+ MockDistributedFileSystem fs = mock(MockDistributedFileSystem.class);
+ when(fs.getErasureCodingPolicy(any())).thenReturn(ReplicationPolicy.OTHER);
+
+ ReplicationPolicy mockPolicy = mock(ReplicationPolicy.class);
+ SystemErasureCodingPolicies.setSystemPolicy(mockPolicy);
+ when(mockPolicy.getName()).thenThrow(createNoSuchMethodException());
+ ECPolicyDisabler.Result result = ECPolicyDisabler.check(fs, null);
+ assertEquals("result is expected", Result.NO_SUCH_METHOD, result);
+ verify(fs).getErasureCodingPolicy(any());
+ verifyNoMoreInteractions(fs);
+ }
+
+ @Test
+ public void testServerNotSupportsSetErasureCodingPolicyMethod() {
+ MockDistributedFileSystem fs = mock(MockDistributedFileSystem.class);
+ when(fs.getErasureCodingPolicy(any())).thenReturn(ReplicationPolicy.OTHER);
+ Mockito.doThrow(createNoSuchMethodException()).when(fs).setErasureCodingPolicy(any(), any());
+ ECPolicyDisabler.Result result = ECPolicyDisabler.check(fs, null);
+ assertEquals("result is expected", Result.NO_SUCH_METHOD, result);
+ verify(fs).getErasureCodingPolicy(any());
+ verify(fs).setErasureCodingPolicy(any(), eq("DEFAULT"));
+ verifyNoMoreInteractions(fs);
+ }
+
+ @Test
public void testOtherRuntimeExceptionThrown() {
MockDistributedFileSystem fs = mock(MockDistributedFileSystem.class);
when(fs.getErasureCodingPolicy(any())).thenReturn(ReplicationPolicy.OTHER);
