You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@oozie.apache.org by ge...@apache.org on 2017/07/12 07:29:01 UTC
[1/2] oozie git commit: OOZIE-2854 Oozie should handle transient
database problems (andras.piros via gezapeti)
Repository: oozie
Updated Branches:
refs/heads/master a436d9829 -> a54f7c20d
http://git-wip-us.apache.org/repos/asf/oozie/blob/a54f7c20/core/src/main/resources/oozie-default.xml
----------------------------------------------------------------------
diff --git a/core/src/main/resources/oozie-default.xml b/core/src/main/resources/oozie-default.xml
index 5629a89..832bbe1 100644
--- a/core/src/main/resources/oozie-default.xml
+++ b/core/src/main/resources/oozie-default.xml
@@ -1490,9 +1490,13 @@ will be the requeue interval for the actions which are waiting for a long time w
<property>
<name>oozie.service.JPAService.connection.data.source</name>
- <value>org.apache.commons.dbcp.BasicDataSource</value>
+ <value>org.apache.oozie.util.db.BasicDataSourceWrapper</value>
<description>
- DataSource to be used for connection pooling.
+ DataSource to be used for connection pooling. If you want the property
+ openJpa.connectionProperties="DriverClassName=..." to have a real effect, set this to
+ org.apache.oozie.util.db.BasicDataSourceWrapper.
+ A DBCP bug (https://issues.apache.org/jira/browse/DBCP-333) prevents otherwise the JDBC driver
+ setting to have a real effect while using custom class loader.
</description>
</property>
@@ -1563,6 +1567,31 @@ will be the requeue interval for the actions which are waiting for a long time w
</description>
</property>
+ <property>
+ <name>oozie.service.JPAService.retry.initial-wait-time.ms</name>
+ <value>100</value>
+ <description>
+ Initial wait time in milliseconds between the first failed database operation and the re-attempted operation. The wait
+ time is doubled at each retry.
+ </description>
+ </property>
+
+ <property>
+ <name>oozie.service.JPAService.retry.maximum-wait-time.ms</name>
+ <value>30000</value>
+ <description>
+ Maximum wait time between database retry attempts.
+ </description>
+ </property>
+
+ <property>
+ <name>oozie.service.JPAService.retry.max-retries</name>
+ <value>10</value>
+ <description>
+ Maximum number of retries for a failed database operation.
+ </description>
+ </property>
+
<!-- SchemaService -->
<property>
http://git-wip-us.apache.org/repos/asf/oozie/blob/a54f7c20/core/src/main/resources/oozie-log4j.properties
----------------------------------------------------------------------
diff --git a/core/src/main/resources/oozie-log4j.properties b/core/src/main/resources/oozie-log4j.properties
index c065f3c..ba8d7b9 100644
--- a/core/src/main/resources/oozie-log4j.properties
+++ b/core/src/main/resources/oozie-log4j.properties
@@ -59,3 +59,4 @@ log4j.logger.org.apache.oozie.sla=DEBUG, test
log4j.logger.org.apache.hadoop=INFO, test
log4j.logger.org.mortbay=INFO, test
log4j.logger.org.hsqldb=INFO, test
+log4j.logger.org.apache.oozie.util.db=INFO, test
http://git-wip-us.apache.org/repos/asf/oozie/blob/a54f7c20/core/src/test/java/org/apache/oozie/command/SkipCommitFaultInjection.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/oozie/command/SkipCommitFaultInjection.java b/core/src/test/java/org/apache/oozie/command/SkipCommitFaultInjection.java
deleted file mode 100644
index 158fbfa..0000000
--- a/core/src/test/java/org/apache/oozie/command/SkipCommitFaultInjection.java
+++ /dev/null
@@ -1,41 +0,0 @@
-/**
- * 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.oozie.command;
-
-import org.apache.oozie.FaultInjection;
-
-public class SkipCommitFaultInjection extends FaultInjection {
-
- public static final String ACTION_FAILOVER_FAULT_INJECTION = "oozie.fault.injection.action.failover";
-
- private static boolean ACTIVE = false;
-
- public boolean activate() {
- ACTIVE = Boolean.parseBoolean(System.getProperty(ACTION_FAILOVER_FAULT_INJECTION, "false"));
- return ACTIVE;
- }
-
- public void deactivate() {
- ACTIVE = false;
- }
-
- public boolean isActive() {
- return ACTIVE;
- }
-}
http://git-wip-us.apache.org/repos/asf/oozie/blob/a54f7c20/core/src/test/java/org/apache/oozie/test/XTestCase.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/oozie/test/XTestCase.java b/core/src/test/java/org/apache/oozie/test/XTestCase.java
index 2816e5d..0ef4a4b 100644
--- a/core/src/test/java/org/apache/oozie/test/XTestCase.java
+++ b/core/src/test/java/org/apache/oozie/test/XTestCase.java
@@ -41,7 +41,8 @@ import java.util.concurrent.atomic.AtomicLong;
import javax.persistence.EntityManager;
import javax.persistence.FlushModeType;
-import javax.persistence.Query;
+import javax.persistence.PersistenceException;
+import javax.persistence.TypedQuery;
import junit.framework.TestCase;
@@ -96,6 +97,8 @@ import org.apache.oozie.util.IOUtils;
import org.apache.oozie.util.ParamChecker;
import org.apache.oozie.util.XConfiguration;
import org.apache.oozie.util.XLog;
+import org.apache.openjpa.persistence.ArgumentException;
+import org.apache.openjpa.persistence.RollbackException;
/**
* Base JUnit <code>TestCase</code> subclass used by all Oozie testcases.
@@ -832,71 +835,29 @@ public abstract class XTestCase extends TestCase {
entityManager.setFlushMode(FlushModeType.COMMIT);
entityManager.getTransaction().begin();
- Query q = entityManager.createNamedQuery("GET_WORKFLOWS");
- List<WorkflowJobBean> wfjBeans = q.getResultList();
- int wfjSize = wfjBeans.size();
- for (WorkflowJobBean w : wfjBeans) {
- entityManager.remove(w);
- }
-
- q = entityManager.createNamedQuery("GET_ACTIONS");
- List<WorkflowActionBean> wfaBeans = q.getResultList();
- int wfaSize = wfaBeans.size();
- for (WorkflowActionBean w : wfaBeans) {
- entityManager.remove(w);
- }
-
- q = entityManager.createNamedQuery("GET_COORD_JOBS");
- List<CoordinatorJobBean> cojBeans = q.getResultList();
- int cojSize = cojBeans.size();
- for (CoordinatorJobBean w : cojBeans) {
- entityManager.remove(w);
- }
-
- q = entityManager.createNamedQuery("GET_COORD_ACTIONS");
- List<CoordinatorActionBean> coaBeans = q.getResultList();
- int coaSize = coaBeans.size();
- for (CoordinatorActionBean w : coaBeans) {
- entityManager.remove(w);
- }
-
- q = entityManager.createNamedQuery("GET_BUNDLE_JOBS");
- List<BundleJobBean> bjBeans = q.getResultList();
- int bjSize = bjBeans.size();
- for (BundleJobBean w : bjBeans) {
- entityManager.remove(w);
- }
-
- q = entityManager.createNamedQuery("GET_BUNDLE_ACTIONS");
- List<BundleActionBean> baBeans = q.getResultList();
- int baSize = baBeans.size();
- for (BundleActionBean w : baBeans) {
- entityManager.remove(w);
- }
+ final int wfjSize = getCountAndRemoveAll(entityManager, "GET_WORKFLOWS", WorkflowJobBean.class);
+ final int wfaSize = getCountAndRemoveAll(entityManager, "GET_ACTIONS", WorkflowActionBean.class);
+ final int cojSize = getCountAndRemoveAll(entityManager, "GET_COORD_JOBS", CoordinatorJobBean.class);
+ final int coaSize = getCountAndRemoveAll(entityManager, "GET_COORD_ACTIONS", CoordinatorActionBean.class);
+ final int bjSize = getCountAndRemoveAll(entityManager, "GET_BUNDLE_JOBS", BundleJobBean.class);
+ final int baSize = getCountAndRemoveAll(entityManager, "GET_BUNDLE_ACTIONS", BundleActionBean.class);
+ final int slaSize = getCountAndRemoveAll(entityManager, "GET_SLA_EVENTS", SLAEventBean.class);
+ final int slaRegSize = getCountAndRemoveAll(entityManager, "GET_ACTIONS", SLARegistrationBean.class);
+ final int ssSize = getCountAndRemoveAll(entityManager, "GET_SLA_SUMMARY_ALL", SLASummaryBean.class);
- q = entityManager.createNamedQuery("GET_SLA_EVENTS");
- List<SLAEventBean> slaBeans = q.getResultList();
- int slaSize = slaBeans.size();
- for (SLAEventBean w : slaBeans) {
- entityManager.remove(w);
- }
+ try {
+ if (entityManager.getTransaction().isActive()) {
+ entityManager.getTransaction().commit();
+ }
- q = entityManager.createQuery("select OBJECT(w) from SLARegistrationBean w");
- List<SLARegistrationBean> slaRegBeans = q.getResultList();
- int slaRegSize = slaRegBeans.size();
- for (SLARegistrationBean w : slaRegBeans) {
- entityManager.remove(w);
+ if (entityManager.isOpen()) {
+ entityManager.close();
+ }
}
-
- q = entityManager.createQuery("select OBJECT(w) from SLASummaryBean w");
- List<SLASummaryBean> sdBeans = q.getResultList();
- int ssSize = sdBeans.size();
- for (SLASummaryBean w : sdBeans) {
- entityManager.remove(w);
+ catch (final RollbackException e) {
+ log.warn("Cannot commit current transaction. [e.message={0}]", e.getMessage());
}
- entityManager.getTransaction().commit();
- entityManager.close();
log.info(wfjSize + " entries in WF_JOBS removed from DB!");
log.info(wfaSize + " entries in WF_ACTIONS removed from DB!");
log.info(cojSize + " entries in COORD_JOBS removed from DB!");
@@ -906,7 +867,30 @@ public abstract class XTestCase extends TestCase {
log.info(slaSize + " entries in SLA_EVENTS removed from DB!");
log.info(slaRegSize + " entries in SLA_REGISTRATION removed from DB!");
log.info(ssSize + " entries in SLA_SUMMARY removed from DB!");
+ }
+ private <E> int getCountAndRemoveAll(final EntityManager entityManager,
+ final String queryName,
+ final Class<E> entityClass) {
+ try {
+ final TypedQuery<E> getAllQuery = entityManager.createNamedQuery(queryName, entityClass);
+ final List<E> allEntities = getAllQuery.getResultList();
+ final int entitiesCount = allEntities.size();
+
+ for (final E w : allEntities) {
+ entityManager.remove(w);
+ }
+
+ return entitiesCount;
+ } catch (final RollbackException e) {
+ log.warn("Cannot get count or remove all entities. [queryName={0};entityClass.name={1}]",
+ queryName, entityClass.getName());
+ return 0;
+ } catch (final PersistenceException | ArgumentException e) {
+ log.warn("Cannot get count or remove all entities. [queryName={0};entityClass.name={1}]",
+ queryName, entityClass.getName());
+ return 0;
+ }
}
private static MiniDFSCluster dfsCluster = null;
http://git-wip-us.apache.org/repos/asf/oozie/blob/a54f7c20/core/src/test/java/org/apache/oozie/util/db/TestOozieDmlStatementPredicate.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/oozie/util/db/TestOozieDmlStatementPredicate.java b/core/src/test/java/org/apache/oozie/util/db/TestOozieDmlStatementPredicate.java
new file mode 100644
index 0000000..c2d4b15
--- /dev/null
+++ b/core/src/test/java/org/apache/oozie/util/db/TestOozieDmlStatementPredicate.java
@@ -0,0 +1,63 @@
+/**
+ * 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.oozie.util.db;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class TestOozieDmlStatementPredicate {
+
+ private final FailingConnectionWrapper.OozieDmlStatementPredicate statementPredicate =
+ new FailingConnectionWrapper.OozieDmlStatementPredicate();
+
+ @Test
+ public void testDmlNotOozieTableDoesNotApply() {
+ assertFalse("DML statement but not of an Oozie table",
+ statementPredicate.apply("SELECT * FROM wellhello"));
+ }
+
+ @Test
+ public void testNotDmlOozieTableDoesNotApply() {
+ assertFalse("not a DML statement but of an Oozie table",
+ statementPredicate.apply("CREATE TABLE WF_JOBS"));
+ }
+
+ @Test
+ public void testNotDmlNotOozieTableDoesNotApply() {
+ assertFalse("not a DML statement and not of an Oozie table",
+ statementPredicate.apply("CREATE TABLE wellhello"));
+ }
+
+ @Test
+ public void testDmlAndOozieTableAppliesIgnoreCase() {
+ assertTrue("a DML statement and of an Oozie table",
+ statementPredicate.apply("SELECT * FROM wf_jobs"));
+
+ assertTrue("a DML statement and of an Oozie table",
+ statementPredicate.apply("insert into WF_JOBS"));
+
+ assertTrue("a DML statement and of an Oozie table",
+ statementPredicate.apply("update wf_jobs"));
+
+ assertTrue("a DML statement and of an Oozie table",
+ statementPredicate.apply("DELETE FROM WF_JOBS"));
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/oozie/blob/a54f7c20/core/src/test/java/org/apache/oozie/util/db/TestOperationRetryHandler.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/oozie/util/db/TestOperationRetryHandler.java b/core/src/test/java/org/apache/oozie/util/db/TestOperationRetryHandler.java
new file mode 100644
index 0000000..ae94199
--- /dev/null
+++ b/core/src/test/java/org/apache/oozie/util/db/TestOperationRetryHandler.java
@@ -0,0 +1,218 @@
+/**
+ * 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.oozie.util.db;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.BDDMockito.willAnswer;
+import static org.mockito.Mockito.mock;
+
+import java.util.concurrent.Callable;
+
+import org.apache.commons.lang.mutable.MutableInt;
+import org.junit.Test;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+
+import com.google.common.base.Predicate;
+
+/**
+ * Conventions used in the tests:
+ * <ul>
+ * <li>{@code RuntimeException} indicates an SQL error (network failure) where we should retry</li>
+ * <li>{@code IllegalStateException} is a different SQL error where we should not retry</li>
+ * </ul>
+ */
+public class TestOperationRetryHandler {
+
+ private final Predicate<Throwable> testRetryPredicate = new Predicate<Throwable>() {
+ @Override
+ public boolean apply(Throwable input) {
+ return input.getClass() == RuntimeException.class;
+ }
+ };
+
+ private final OperationRetryHandler retryHandler = new OperationRetryHandler(3, 1, 1, testRetryPredicate);
+
+ @Test
+ public void testWhenConfiguredNotToRetryThenTriesOnce() throws Exception {
+ tryOnceAndAssert(new OperationRetryHandler(0, 1, 100, testRetryPredicate));
+ tryOnceAndAssert(new OperationRetryHandler(10, 1, 0, testRetryPredicate));
+ tryOnceAndAssert(new OperationRetryHandler(0, 1, 0, testRetryPredicate));
+ }
+
+ private void tryOnceAndAssert(final OperationRetryHandler tryingOnceRetryHandler) throws Exception {
+ @SuppressWarnings("unchecked")
+ final Callable<String> operation = mock(Callable.class);
+ final MutableInt callCount = new MutableInt(0);
+
+ willAnswer(new Answer<Void>() {
+ @Override
+ public Void answer(InvocationOnMock invocation) throws Throwable {
+ callCount.increment();
+ return null;
+ }
+ }).given(operation).call();
+
+ tryingOnceRetryHandler.executeWithRetry(operation);
+
+ assertCallCount(1, callCount);
+ assertNoRetryAttemptsAreInProgressOrExhausted();
+ }
+
+ @Test
+ public void testNoRetry() throws Exception {
+ @SuppressWarnings("unchecked")
+ Callable<String> operation = mock(Callable.class);
+ given(operation.call()).willReturn("dummy");
+
+ retryHandler.executeWithRetry(operation);
+ }
+
+ @Test
+ public void testRetriesOnFailure() throws Exception {
+ @SuppressWarnings("unchecked")
+ Callable<String> operation = mock(Callable.class);
+ final MutableInt callCount = new MutableInt(0);
+
+ willAnswer(new Answer<Void>() {
+ @Override
+ public Void answer(InvocationOnMock invocation) throws Throwable {
+ callCount.increment();
+ throw new RuntimeException();
+ }
+ }).given(operation).call();
+
+ boolean exceptionThrown = false;
+ try {
+ retryHandler.executeWithRetry(operation);
+ } catch (RuntimeException e) {
+ exceptionThrown = true;
+ }
+
+ assertTrue("Exception was not thrown", exceptionThrown);
+ assertCallCount(3, callCount);
+ assertNoRetryAttemptsAreInProgressOrExhausted();
+ }
+
+ @Test
+ public void testRetriesOnFailureRecovery() throws Exception {
+ @SuppressWarnings("unchecked")
+ Callable<String> operation = mock(Callable.class);
+ final MutableInt callCount = new MutableInt(0);
+
+ willAnswer(new Answer<Void>() {
+ @Override
+ public Void answer(InvocationOnMock invocation) throws Throwable {
+ callCount.increment();
+ if (callCount.intValue() < 2) {
+ throw new RuntimeException();
+ }
+ return null;
+ }
+ }).given(operation).call();
+
+ retryHandler.executeWithRetry(operation);
+
+ assertCallCount(2, callCount);
+ assertNoRetryAttemptsAreInProgressOrExhausted();
+ }
+
+ @Test
+ public void testNoRetriesOnNonSQLError() throws Exception {
+ @SuppressWarnings("unchecked")
+ Callable<String> operation = mock(Callable.class);
+ final MutableInt callCount = new MutableInt(0);
+
+ willAnswer(new Answer<Void>() {
+ @Override
+ public Void answer(InvocationOnMock invocation) throws Throwable {
+ callCount.increment();
+ throw new IllegalStateException();
+ }
+ }).given(operation).call();
+
+ boolean exceptionThrown = false;
+ try {
+ retryHandler.executeWithRetry(operation);
+ } catch (IllegalStateException e) {
+ exceptionThrown = true;
+ }
+
+ assertTrue("Exception was not thrown", exceptionThrown);
+ assertCallCount(1, callCount);
+ assertNoRetryAttemptsAreInProgressOrExhausted();
+ }
+
+ @Test
+ public void testEmbeddedRetryHandlersWhenInnerHandlerThrowsSQLError() throws Exception {
+ @SuppressWarnings("unchecked")
+ final Callable<String> innerOperation = mock(Callable.class);
+ final MutableInt callCount = new MutableInt(0);
+
+ // internal operation which will be retried
+ willAnswer(new Answer<Void>() {
+ @Override
+ public Void answer(InvocationOnMock invocation) throws Throwable {
+ callCount.increment();
+ throw new RuntimeException();
+ }
+ }).given(innerOperation).call();
+
+ @SuppressWarnings("unchecked")
+ Callable<String> outerOperation = mock(Callable.class);
+
+ // the outer operation which calls the inner one and this is not
+ // supposed to be retried
+ willAnswer(new Answer<Void>() {
+ @Override
+ public Void answer(InvocationOnMock invocation) throws Throwable {
+ try {
+ retryHandler.executeWithRetry(innerOperation);
+ } catch (Exception e) {
+ throw new RuntimeException();
+ }
+ return null;
+ }
+ }).given(outerOperation).call();
+
+ boolean exceptionThrown = false;
+ try {
+ retryHandler.executeWithRetry(outerOperation);
+ } catch (Exception e) {
+ exceptionThrown = true;
+ }
+
+ assertTrue("Exception was not thrown", exceptionThrown);
+ assertCallCount(3, callCount);
+ assertNoRetryAttemptsAreInProgressOrExhausted();
+ }
+
+ private void assertCallCount(int expected, MutableInt callCount) {
+ assertEquals("Number of retries", expected, callCount.intValue());
+ }
+
+ private void assertNoRetryAttemptsAreInProgressOrExhausted() {
+ assertEquals("Nesting level", 0,
+ OperationRetryHandler.RETRY_ATTEMPT_STATE.getInProgressCount());
+ assertEquals("Retries performed", false,
+ OperationRetryHandler.RETRY_ATTEMPT_STATE.isExhausted());
+ }
+}
http://git-wip-us.apache.org/repos/asf/oozie/blob/a54f7c20/core/src/test/java/org/apache/oozie/util/db/TestPersistenceExceptionSubclassFilterRetryPredicate.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/oozie/util/db/TestPersistenceExceptionSubclassFilterRetryPredicate.java b/core/src/test/java/org/apache/oozie/util/db/TestPersistenceExceptionSubclassFilterRetryPredicate.java
new file mode 100644
index 0000000..76d2edc
--- /dev/null
+++ b/core/src/test/java/org/apache/oozie/util/db/TestPersistenceExceptionSubclassFilterRetryPredicate.java
@@ -0,0 +1,98 @@
+/**
+ * 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.oozie.util.db;
+
+import org.apache.oozie.ErrorCode;
+import org.apache.oozie.XException;
+import org.apache.oozie.executor.jpa.JPAExecutorException;
+import org.junit.Test;
+
+import javax.persistence.EntityExistsException;
+import javax.persistence.EntityNotFoundException;
+import javax.persistence.LockTimeoutException;
+import javax.persistence.NoResultException;
+import javax.persistence.NonUniqueResultException;
+import javax.persistence.OptimisticLockException;
+import javax.persistence.PersistenceException;
+import javax.persistence.PessimisticLockException;
+import javax.persistence.QueryTimeoutException;
+import javax.persistence.RollbackException;
+import javax.persistence.TransactionRequiredException;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class TestPersistenceExceptionSubclassFilterRetryPredicate {
+ private final PersistenceExceptionSubclassFilterRetryPredicate predicate =
+ new PersistenceExceptionSubclassFilterRetryPredicate();
+
+ @Test
+ public void testFilteredJPAExceptions() {
+ assertFalse(predicate.apply(new EntityExistsException()));
+ assertFalse(predicate.apply(new EntityNotFoundException()));
+ assertFalse(predicate.apply(new LockTimeoutException()));
+ assertFalse(predicate.apply(new NoResultException()));
+ assertFalse(predicate.apply(new NonUniqueResultException()));
+ assertFalse(predicate.apply(new OptimisticLockException()));
+ assertFalse(predicate.apply(new PessimisticLockException()));
+ assertFalse(predicate.apply(new QueryTimeoutException()));
+ assertFalse(predicate.apply(new TransactionRequiredException()));
+ }
+
+ @Test
+ public void testNotFilteredJPAExceptions() {
+ assertTrue(predicate.apply(new RollbackException()));
+ assertTrue(predicate.apply(new PersistenceException()));
+ }
+
+ @Test
+ public void testNonJPAExceptions() {
+ assertTrue(predicate.apply(new RuntimeException()));
+ assertTrue(predicate.apply(new Exception()));
+ }
+
+ @Test
+ public void testNestedFilteredJPAExceptions() {
+ assertFalse(predicate.apply(wrapCause(new EntityExistsException())));
+ assertFalse(predicate.apply(wrapCause(new EntityNotFoundException())));
+ assertFalse(predicate.apply(wrapCause(new LockTimeoutException())));
+ assertFalse(predicate.apply(wrapCause(new NoResultException())));
+ assertFalse(predicate.apply(wrapCause(new NonUniqueResultException())));
+ assertFalse(predicate.apply(wrapCause(new OptimisticLockException())));
+ assertFalse(predicate.apply(wrapCause(new PessimisticLockException())));
+ assertFalse(predicate.apply(wrapCause(new QueryTimeoutException())));
+ assertFalse(predicate.apply(wrapCause(new TransactionRequiredException())));
+ }
+
+ @Test
+ public void testNestedNotFilteredJPAExceptions() {
+ assertTrue(predicate.apply(wrapCause(new RollbackException())));
+ assertTrue(predicate.apply(wrapCause(new PersistenceException())));
+ }
+
+ @Test
+ public void testNestedNonJPAExceptions() {
+ assertTrue(predicate.apply(wrapCause(new RuntimeException())));
+ assertTrue(predicate.apply(wrapCause(new Exception())));
+ }
+
+ private JPAExecutorException wrapCause(final Throwable cause) {
+ return new JPAExecutorException(new XException(ErrorCode.E0603, new PersistenceException(cause)));
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/oozie/blob/a54f7c20/core/src/test/java/org/apache/oozie/util/db/TestRetryAttemptState.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/oozie/util/db/TestRetryAttemptState.java b/core/src/test/java/org/apache/oozie/util/db/TestRetryAttemptState.java
new file mode 100644
index 0000000..0158538
--- /dev/null
+++ b/core/src/test/java/org/apache/oozie/util/db/TestRetryAttemptState.java
@@ -0,0 +1,132 @@
+/**
+ * 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.oozie.util.db;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class TestRetryAttemptState {
+ private final RetryAttemptState state = new RetryAttemptState();
+
+ @Test
+ public void testOneAttemptIsInProgressAndAttemptsAreNotExhausted() {
+ state.signalStart();
+
+ assertFalse("retry attempts exhausted", state.isExhausted());
+ assertEquals("retry attempt count", 1, state.getInProgressCount());
+ }
+
+ @Test
+ public void testNoAttemptIsInProgressAndAttemptsAreNotExhausted() {
+ state.signalStart();
+ state.signalEnd();
+
+ assertFalse("retry attempts exhausted", state.isExhausted());
+ assertEquals("retry attempt count", 0, state.getInProgressCount());
+ }
+
+ @Test
+ public void testExhaustedFromOneThreadRemainsNotStartedFromAnotherThread() throws InterruptedException {
+ final Runnable exhaustingAttempt = new ExhaustingAttempt();
+
+ final Thread exhaustingThread = new Thread(exhaustingAttempt);
+
+ exhaustingThread.start();
+
+ exhaustingThread.join();
+
+ assertFalse("retry attempts exhausted", state.isExhausted());
+ assertEquals("retry attempt count", 0, state.getInProgressCount());
+ }
+
+ @Test
+ public void testNestedAttemptsFinishSuccessfullyWhenNotExhausted() throws InterruptedException {
+ final NestedAttempt inner = new NestedAttempt(false);
+ final NestedAttempt outer = new NestedAttempt(inner, false);
+
+ outer.run();
+
+ assertFalse("retry attempts exhausted", state.isExhausted());
+ assertEquals("retry attempt count", 0, state.getInProgressCount());
+ }
+
+ @Test
+ public void testOuterAttemptRemainsUnfinishedWhenInnerExhausted() throws InterruptedException {
+ final NestedAttempt inner = new NestedAttempt(true);
+ final NestedAttempt outer = new NestedAttempt(inner, false);
+
+ outer.run();
+
+ assertTrue("retry attempts exhausted", state.isExhausted());
+ assertEquals("retry attempt count", 1, state.getInProgressCount());
+ }
+
+ @Test
+ public void testInnerAttemptRemainsUnfinishedWhenOuterExhausted() throws InterruptedException {
+ final NestedAttempt inner = new NestedAttempt(false);
+ final NestedAttempt outer = new NestedAttempt(inner, true);
+
+ outer.run();
+
+ assertTrue("retry attempts exhausted", state.isExhausted());
+ assertEquals("retry attempt count", 1, state.getInProgressCount());
+ }
+
+ private class ExhaustingAttempt implements Runnable {
+ @Override
+ public void run() {
+ state.signalStart();
+
+ state.signalExhausted();
+ }
+ }
+
+ private class NestedAttempt implements Runnable {
+ private final NestedAttempt nestedAttempt;
+ private final boolean shouldExhaust;
+
+ NestedAttempt(final boolean shouldExhaust) {
+ this(null, shouldExhaust);
+ }
+
+ NestedAttempt(final NestedAttempt nestedAttempt, final boolean shouldExhaust) {
+ this.nestedAttempt = nestedAttempt;
+ this.shouldExhaust = shouldExhaust;
+ }
+
+ @Override
+ public void run() {
+ state.signalStart();
+
+ if (nestedAttempt != null) {
+ nestedAttempt.run();
+ }
+
+ if (shouldExhaust) {
+ state.signalExhausted();
+ }
+ else {
+ state.signalEnd();
+ }
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/oozie/blob/a54f7c20/docs/src/site/twiki/AG_Install.twiki
----------------------------------------------------------------------
diff --git a/docs/src/site/twiki/AG_Install.twiki b/docs/src/site/twiki/AG_Install.twiki
index 2125442..3509266 100644
--- a/docs/src/site/twiki/AG_Install.twiki
+++ b/docs/src/site/twiki/AG_Install.twiki
@@ -198,6 +198,37 @@ following configuration properties in the oozie-site.xml:
oozie.service.JPAService.pool.max.active.conn=10
</verbatim>
+If you are interested in fine tuning how Oozie can retry database operations on failing database connectivity or errors, you can
+set following properties to other values. Here are the default ones:
+
+<verbatim>
+ oozie.service.JPAService.retry.initial-wait-time.ms=100
+ oozie.service.JPAService.retry.maximum-wait-time.ms=30000
+ oozie.service.JPAService.retry.max-retries=10
+</verbatim>
+
+If you set either =oozie.service.JPAService.retry.max-retries= or =oozie.service.JPAService.retry.maximum-wait-time.ms= to =0=,
+no retry attempts will be made on any database connectivity issues. Exact settings for these properties depend also on how much load
+is on Oozie regarding workflow and coordinator jobs.
+
+The database operation retry functionality kicks in when there is a =javax.persistence.PersistenceException= those root cause is not
+part of the normal everyday operation - filtered against a blacklist consisting of descendants like =NoSuchResultException=,
+=NonUniqueResultException=, and the like. This way Oozie won't retry database operations on errors that are more related to the
+current query, or otherwise part of the everyday life. This way it's ensured that this blacklist is database agnostic.
+
+It has been tested with a MySQL / failing every minute 10 seconds / an Oozie coordinator job of an Oozie workflow consisting of four
+workflow actions (some of them are asynchronous). On this setup Oozie was recovering after each and every database outage.
+
+To set up such a failing MySQL scenario following has to be performed:
+
+ * Set =oozie.service.JPAService.connection.data.source= to =org.apache.oozie.util.db.BasicDataSourceWrapper=
+ within =oozie-site.xml=
+ * Set =oozie.service.JPAService.jdbc.driver= to =org.apache.oozie.util.db.FailingMySQLDriverWrapper= within =oozie-site.xml=
+ * Restart Oozie server
+ * Submit / start some workflows, coordinators etc.
+ * See how Oozie is retrying on injected database errors by looking at the Oozie server logs, grepping =JPAException= instances
+ with following message prefix: <verbatim>Deliberately failing to prepare statement.</verbatim>
+
---++ Database Migration
Oozie provides an easy way to switch between databases without losing any data. Oozie servers should be stopped during the
http://git-wip-us.apache.org/repos/asf/oozie/blob/a54f7c20/minitest/pom.xml
----------------------------------------------------------------------
diff --git a/minitest/pom.xml b/minitest/pom.xml
index 9515284..10a89b0 100644
--- a/minitest/pom.xml
+++ b/minitest/pom.xml
@@ -48,13 +48,13 @@
<dependency>
<groupId>org.codehaus.jackson</groupId>
<artifactId>jackson-mapper-asl</artifactId>
- <version>1.5.2</version>
+ <version>1.8.8</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.codehaus.jackson</groupId>
<artifactId>jackson-core-asl</artifactId>
- <version>1.5.2</version>
+ <version>1.8.8</version>
<scope>test</scope>
</dependency>
<dependency>
http://git-wip-us.apache.org/repos/asf/oozie/blob/a54f7c20/minitest/src/test/java/org/apache/oozie/test/TestParallelJPAOperationRetries.java
----------------------------------------------------------------------
diff --git a/minitest/src/test/java/org/apache/oozie/test/TestParallelJPAOperationRetries.java b/minitest/src/test/java/org/apache/oozie/test/TestParallelJPAOperationRetries.java
new file mode 100644
index 0000000..5442130
--- /dev/null
+++ b/minitest/src/test/java/org/apache/oozie/test/TestParallelJPAOperationRetries.java
@@ -0,0 +1,144 @@
+/**
+ * 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.oozie.test;
+
+import org.apache.oozie.WorkflowActionBean;
+import org.apache.oozie.service.JPAService;
+import org.apache.oozie.service.Services;
+import org.apache.oozie.service.XLogService;
+import org.apache.oozie.util.XLog;
+import org.apache.oozie.util.db.FailingHSQLDBDriverWrapper;
+
+import javax.persistence.EntityManager;
+import javax.persistence.TypedQuery;
+import java.sql.SQLException;
+import java.sql.Timestamp;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * To test that when a JDBC driver fails, {@link JPAService} continues working:
+ * <ul>
+ * <li>use the {@link java.sql.Connection} wrapper that is used by the wrapper extending {@link org.hsqldb.jdbcDriver},
+ * and set the system property {@link FailingHSQLDBDriverWrapper#USE_FAILING_DRIVER}</li>
+ * <li>initialize {@link JPAService} and a JPA {@link javax.persistence.EntityManagerFactory}</li>
+ * <li>issue a decent number of JPA queries w/ appropriate parallelism via an {@link java.util.concurrent.Executor}</li>
+ * <li>an {@link TypedQuery#getResultList()} resulting in a {@code SELECT … FROM WF_JOBS WHERE ...} Oozie database read</li>
+ * <li>followed by a {@link TypedQuery#executeUpdate()} resulting in an {@code UPDATE WF_ACTIONS SET … WHERE …} Oozie database
+ * write</li>
+ * </ul>
+ */
+public class TestParallelJPAOperationRetries extends MiniOozieTestCase {
+ private static final XLog LOG = XLog.getLog(TestParallelJPAOperationRetries.class);
+ private static final String ORIGINAL_LOG4J_FILE = System.getProperty(XLogService.LOG4J_FILE);
+
+ private volatile Exception executorException;
+
+ @Override
+ protected void setUp() throws Exception {
+ executorException = null;
+ System.setProperty(XLogService.LOG4J_FILE, "oozie-log4j.properties");
+ System.setProperty(FailingHSQLDBDriverWrapper.USE_FAILING_DRIVER, Boolean.TRUE.toString());
+
+ super.setUp();
+ }
+
+ @Override
+ protected void tearDown() throws Exception {
+ super.tearDown();
+
+ if (ORIGINAL_LOG4J_FILE != null) {
+ System.setProperty(XLogService.LOG4J_FILE, ORIGINAL_LOG4J_FILE);
+ }
+ }
+
+ public void testParallelJPAOperationsOnWorkflowBeansRetryAndSucceed() throws Exception {
+ final ExecutorService taskExecutor = Executors.newFixedThreadPool(4);
+ final int totalTaskCount = 10;
+ final CountDownLatch countDownLatch = new CountDownLatch(totalTaskCount);
+
+ for (int ixTask = 1; ixTask <= totalTaskCount; ixTask++) {
+ final int taskCount = ixTask;
+ taskExecutor.execute(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ LOG.debug("Task #{0} started.", taskCount);
+
+ getResultListAndExecuteUpdateOnWorkflowBeans();
+
+ LOG.debug("Task #{0} finished.", taskCount);
+ }
+ catch (final SQLException e) {
+ executorException = e;
+ }
+ finally {
+ countDownLatch.countDown();
+ }
+ }
+ });
+ }
+
+ countDownLatch.await();
+
+ if (executorException != null) {
+ fail(String.format("Should not get an SQLException while executing SQL operations. [e.message=%s]",
+ executorException.getMessage()));
+ }
+
+ taskExecutor.shutdown();
+
+ try {
+ taskExecutor.awaitTermination(1, TimeUnit.MINUTES);
+ } catch (final InterruptedException e) {
+ fail("Should not get an interrupt while shutting down ExecutorService.");
+ }
+ }
+
+ private void getResultListAndExecuteUpdateOnWorkflowBeans() throws SQLException {
+ final JPAService jpaService = Services.get().get(JPAService.class);
+
+ final int checkAgeSecs = 86_400;
+ final Timestamp ts = new Timestamp(System.currentTimeMillis() - checkAgeSecs * 1000);
+
+ final EntityManager em = jpaService.getEntityManager();
+ final TypedQuery<WorkflowActionBean> q = em.createNamedQuery("GET_RUNNING_ACTIONS", WorkflowActionBean.class);
+ q.setParameter("lastCheckTime", ts);
+ final List<WorkflowActionBean> actions = q.getResultList();
+
+ assertEquals("no WorkflowActionBeans should be present", 0, actions.size());
+
+ if (!em.getTransaction().isActive()) {
+ em.getTransaction().begin();
+ }
+
+ final int rowsAffected = em.createNamedQuery("UPDATE_ACTION_FOR_LAST_CHECKED_TIME")
+ .setParameter("lastCheckTime", 0l)
+ .setParameter("id", "666")
+ .executeUpdate();
+ if (em.getTransaction().isActive()) {
+ em.getTransaction().commit();
+ }
+
+ assertEquals("no rows should be affected when updating WorkflowActionBeans", 0, rowsAffected);
+ }
+}
http://git-wip-us.apache.org/repos/asf/oozie/blob/a54f7c20/minitest/src/test/java/org/apache/oozie/test/TestWorkflow.java
----------------------------------------------------------------------
diff --git a/minitest/src/test/java/org/apache/oozie/test/TestWorkflow.java b/minitest/src/test/java/org/apache/oozie/test/TestWorkflow.java
new file mode 100644
index 0000000..4257b60
--- /dev/null
+++ b/minitest/src/test/java/org/apache/oozie/test/TestWorkflow.java
@@ -0,0 +1,215 @@
+/**
+ * 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.oozie.test;
+
+import com.google.common.base.Strings;
+import org.apache.oozie.action.hadoop.JavaActionExecutor;
+import org.apache.oozie.client.OozieClientException;
+import org.apache.oozie.service.XLogService;
+import org.apache.oozie.client.OozieClient;
+import org.apache.oozie.client.WorkflowJob;
+import org.apache.oozie.local.LocalOozie;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.Reader;
+import java.io.Writer;
+import java.io.OutputStreamWriter;
+import java.util.Date;
+import java.util.Properties;
+
+import static org.junit.Assume.assumeFalse;
+
+/**
+ * {@code MiniOozie} integration test for different workflow kinds.
+ */
+public class TestWorkflow extends MiniOozieTestCase {
+
+ @Override
+ protected void setUp() throws Exception {
+ System.setProperty(XLogService.LOG4J_FILE, "oozie-log4j.properties");
+ super.setUp();
+ }
+
+ @Override
+ protected void tearDown() throws Exception {
+ super.tearDown();
+ }
+
+ public void testWorkflowWithStartAndEndCompletesSuccessfully() throws Exception {
+ final String wfApp = "<workflow-app xmlns='uri:oozie:workflow:0.1' name='test-wf'>" + " <start to='end'/>"
+ + " <end name='end'/>" + "</workflow-app>";
+
+ final FileSystem fs = getFileSystem();
+ final Path appPath = new Path(getFsTestCaseDir(), "app");
+ fs.mkdirs(appPath);
+ fs.mkdirs(new Path(appPath, "lib"));
+
+ final Writer writer = new OutputStreamWriter(fs.create(new Path(appPath, "workflow.xml")));
+ writer.write(wfApp);
+ writer.close();
+
+ final OozieClient wc = LocalOozie.getClient();
+
+ final Properties conf = wc.createConfiguration();
+ conf.setProperty(OozieClient.APP_PATH, new Path(appPath, "workflow.xml").toString());
+ conf.setProperty(OozieClient.USER_NAME, getTestUser());
+
+
+ final String jobId = wc.submit(conf);
+ assertNotNull(jobId);
+
+ WorkflowJob wf = wc.getJobInfo(jobId);
+ assertNotNull(wf);
+ assertEquals(WorkflowJob.Status.PREP, wf.getStatus());
+
+ wc.start(jobId);
+
+ waitFor(1000, new Predicate() {
+ public boolean evaluate() throws Exception {
+ final WorkflowJob wf = wc.getJobInfo(jobId);
+ return wf.getStatus() == WorkflowJob.Status.SUCCEEDED;
+ }
+ });
+
+ wf = wc.getJobInfo(jobId);
+ assertNotNull(wf);
+ assertEquals(WorkflowJob.Status.SUCCEEDED, wf.getStatus());
+ }
+
+ public void testFsDecisionWorkflowCompletesSuccessfully() throws Exception {
+ final String workflowFileName = "fs-decision.xml";
+ final Properties additionalWorkflowProperties = new Properties();
+
+ runWorkflowFromFile(workflowFileName, additionalWorkflowProperties);
+ }
+
+ public void testParallelFsAndShellWorkflowCompletesSuccessfully() throws Exception {
+ final String workflowFileName = "parallel-fs-and-shell.xml";
+ final Properties additionalWorkflowProperties = new Properties();
+ final boolean isEvenSecond = new Date().getTime() % 2 == 0;
+ additionalWorkflowProperties.setProperty("choosePath1", Boolean.toString(isEvenSecond));
+
+ final String envJavaHome = System.getenv("JAVA_HOME");
+ assumeFalse("Environment variable JAVA_HOME has to be set", Strings.isNullOrEmpty(envJavaHome));
+
+ additionalWorkflowProperties.setProperty("oozie.launcher." + JavaActionExecutor.YARN_AM_ENV, envJavaHome);
+
+ runWorkflowFromFile(workflowFileName, additionalWorkflowProperties);
+ }
+
+ private void runWorkflowFromFile(final String workflowFileName, final Properties additionalWorkflowProperties)
+ throws IOException, OozieClientException {
+ final FileSystem fs = getFileSystem();
+ final Path appPath = new Path(getFsTestCaseDir(), "app");
+ fs.mkdirs(appPath);
+ fs.mkdirs(new Path(appPath, "lib"));
+
+ final Reader reader = getResourceAsReader(workflowFileName, -1);
+ final Writer writer = new OutputStreamWriter(fs.create(new Path(appPath, "workflow.xml")));
+ copyCharStream(reader, writer);
+ writer.close();
+ reader.close();
+
+ final Path path = getFsTestCaseDir();
+
+ final OozieClient oozieClient = LocalOozie.getClient();
+
+ final Properties conf = oozieClient.createConfiguration();
+ conf.setProperty(OozieClient.APP_PATH, new Path(appPath, "workflow.xml").toString());
+ conf.setProperty(OozieClient.USER_NAME, getTestUser());
+ conf.setProperty("nameNodeBasePath", path.toString());
+ conf.setProperty("base", path.toUri().getPath());
+ conf.setProperty("nameNode", getNameNodeUri());
+ conf.setProperty("jobTracker", getJobTrackerUri());
+
+ for (final String additionalKey : additionalWorkflowProperties.stringPropertyNames()) {
+ conf.setProperty(additionalKey, additionalWorkflowProperties.getProperty(additionalKey));
+ }
+
+ final String jobId = oozieClient.submit(conf);
+ assertNotNull(jobId);
+
+ WorkflowJob wf = oozieClient.getJobInfo(jobId);
+ assertNotNull(wf);
+ assertEquals(WorkflowJob.Status.PREP, wf.getStatus());
+
+ oozieClient.start(jobId);
+
+ waitFor(15 * 1000, new Predicate() {
+ public boolean evaluate() throws Exception {
+ final WorkflowJob wf = oozieClient.getJobInfo(jobId);
+ return wf.getStatus() == WorkflowJob.Status.SUCCEEDED;
+ }
+ });
+
+ wf = oozieClient.getJobInfo(jobId);
+ assertNotNull(wf);
+ assertEquals(WorkflowJob.Status.SUCCEEDED, wf.getStatus());
+ }
+
+ /**
+ * Return a classpath resource as a stream.
+ * <p/>
+ *
+ * @param path classpath for the resource.
+ * @param maxLen max content length allowed.
+ * @return the stream for the resource.
+ * @throws IOException thrown if the resource could not be read.
+ */
+ private InputStream getResourceAsStream(final String path, final int maxLen) throws IOException {
+ final InputStream is = Thread.currentThread().getContextClassLoader().getResourceAsStream(path);
+ if (is == null) {
+ throw new IllegalArgumentException("resource " + path + " not found");
+ }
+ return is;
+ }
+
+ /**
+ * Return a classpath resource as a reader.
+ * <p/>
+ * It is assumed that the resource is a text resource.
+ *
+ * @param path classpath for the resource.
+ * @param maxLen max content length allowed.
+ * @return the reader for the resource.
+ * @throws IOException thrown if the resource could not be read.
+ */
+ private Reader getResourceAsReader(final String path, final int maxLen) throws IOException {
+ return new InputStreamReader(getResourceAsStream(path, maxLen));
+ }
+
+ /**
+ * Copies an char input stream into an char output stream.
+ *
+ * @param reader reader to copy from.
+ * @param writer writer to copy to.
+ * @throws IOException thrown if the copy failed.
+ */
+ private void copyCharStream(final Reader reader, final Writer writer) throws IOException {
+ final char[] buffer = new char[4096];
+ int read;
+ while ((read = reader.read(buffer)) > -1) {
+ writer.write(buffer, 0, read);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/oozie/blob/a54f7c20/minitest/src/test/java/org/apache/oozie/test/TestWorkflowRetries.java
----------------------------------------------------------------------
diff --git a/minitest/src/test/java/org/apache/oozie/test/TestWorkflowRetries.java b/minitest/src/test/java/org/apache/oozie/test/TestWorkflowRetries.java
new file mode 100644
index 0000000..5ae1bb4
--- /dev/null
+++ b/minitest/src/test/java/org/apache/oozie/test/TestWorkflowRetries.java
@@ -0,0 +1,34 @@
+/**
+ * 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.oozie.test;
+
+import org.apache.oozie.util.db.FailingHSQLDBDriverWrapper;
+
+/**
+ * {@code MiniOozie} integration test for different workflow kinds, using a HSQLDB driver that fails sometimes.
+ */
+public class TestWorkflowRetries extends TestWorkflow {
+
+ @Override
+ protected void setUp() throws Exception {
+ System.setProperty(FailingHSQLDBDriverWrapper.USE_FAILING_DRIVER, Boolean.TRUE.toString());
+
+ super.setUp();
+ }
+}
http://git-wip-us.apache.org/repos/asf/oozie/blob/a54f7c20/minitest/src/test/java/org/apache/oozie/test/WorkflowTest.java
----------------------------------------------------------------------
diff --git a/minitest/src/test/java/org/apache/oozie/test/WorkflowTest.java b/minitest/src/test/java/org/apache/oozie/test/WorkflowTest.java
deleted file mode 100644
index 2845f0a..0000000
--- a/minitest/src/test/java/org/apache/oozie/test/WorkflowTest.java
+++ /dev/null
@@ -1,188 +0,0 @@
-/**
- * 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.oozie.test;
-
-import org.apache.oozie.service.XLogService;
-import org.apache.oozie.test.MiniOozieTestCase;
-import org.apache.oozie.client.OozieClient;
-import org.apache.oozie.client.WorkflowJob;
-import org.apache.oozie.local.LocalOozie;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-
-import java.io.File;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.InputStreamReader;
-import java.io.Reader;
-import java.io.Writer;
-import java.io.OutputStreamWriter;
-import java.util.Properties;
-
-/**
- * MiniOozie unit test
- */
-public class WorkflowTest extends MiniOozieTestCase {
-
- @Override
- protected void setUp() throws Exception {
- System.setProperty(XLogService.LOG4J_FILE, "oozie-log4j.properties");
- super.setUp();
- }
-
- @Override
- protected void tearDown() throws Exception {
- super.tearDown();
- }
-
- public void testWorkflowRun() throws Exception {
- String wfApp = "<workflow-app xmlns='uri:oozie:workflow:0.1' name='test-wf'>" + " <start to='end'/>"
- + " <end name='end'/>" + "</workflow-app>";
-
- FileSystem fs = getFileSystem();
- Path appPath = new Path(getFsTestCaseDir(), "app");
- fs.mkdirs(appPath);
- fs.mkdirs(new Path(appPath, "lib"));
-
- Writer writer = new OutputStreamWriter(fs.create(new Path(appPath, "workflow.xml")));
- writer.write(wfApp);
- writer.close();
-
- final OozieClient wc = LocalOozie.getClient();
-
- Properties conf = wc.createConfiguration();
- conf.setProperty(OozieClient.APP_PATH, new Path(appPath, "workflow.xml").toString());
- conf.setProperty(OozieClient.USER_NAME, getTestUser());
-
-
- final String jobId = wc.submit(conf);
- assertNotNull(jobId);
-
- WorkflowJob wf = wc.getJobInfo(jobId);
- assertNotNull(wf);
- assertEquals(WorkflowJob.Status.PREP, wf.getStatus());
-
- wc.start(jobId);
-
- waitFor(1000, new Predicate() {
- public boolean evaluate() throws Exception {
- WorkflowJob wf = wc.getJobInfo(jobId);
- return wf.getStatus() == WorkflowJob.Status.SUCCEEDED;
- }
- });
-
- wf = wc.getJobInfo(jobId);
- assertNotNull(wf);
- assertEquals(WorkflowJob.Status.SUCCEEDED, wf.getStatus());
-
- }
-
- public void testWorkflowRunFromFile() throws Exception {
- FileSystem fs = getFileSystem();
- Path appPath = new Path(getFsTestCaseDir(), "app");
- fs.mkdirs(appPath);
- fs.mkdirs(new Path(appPath, "lib"));
-
- Reader reader = getResourceAsReader("wf-test.xml", -1);
- Writer writer = new OutputStreamWriter(fs.create(new Path(appPath, "workflow.xml")));
- copyCharStream(reader, writer);
- writer.close();
- reader.close();
-
- Path path = getFsTestCaseDir();
-
- final OozieClient wc = LocalOozie.getClient();
-
- Properties conf = wc.createConfiguration();
- conf.setProperty(OozieClient.APP_PATH, new Path(appPath, "workflow.xml").toString());
- conf.setProperty(OozieClient.USER_NAME, getTestUser());
- conf.setProperty("nnbase", path.toString());
- conf.setProperty("base", path.toUri().getPath());
-
-
-
- final String jobId = wc.submit(conf);
- assertNotNull(jobId);
-
- WorkflowJob wf = wc.getJobInfo(jobId);
- assertNotNull(wf);
- assertEquals(WorkflowJob.Status.PREP, wf.getStatus());
-
- wc.start(jobId);
-
- waitFor(15 * 1000, new Predicate() {
- public boolean evaluate() throws Exception {
- WorkflowJob wf = wc.getJobInfo(jobId);
- return wf.getStatus() == WorkflowJob.Status.SUCCEEDED;
- }
- });
-
- wf = wc.getJobInfo(jobId);
- assertNotNull(wf);
- assertEquals(WorkflowJob.Status.SUCCEEDED, wf.getStatus());
-
- }
-
- /**
- * Return a classpath resource as a stream.
- * <p/>
- *
- * @param path classpath for the resource.
- * @param maxLen max content length allowed.
- * @return the stream for the resource.
- * @throws IOException thrown if the resource could not be read.
- */
- public InputStream getResourceAsStream(String path, int maxLen) throws IOException {
- InputStream is = Thread.currentThread().getContextClassLoader().getResourceAsStream(path);
- if (is == null) {
- throw new IllegalArgumentException("resource " + path + " not found");
- }
- return is;
- }
-
- /**
- * Return a classpath resource as a reader.
- * <p/>
- * It is assumed that the resource is a text resource.
- *
- * @param path classpath for the resource.
- * @param maxLen max content length allowed.
- * @return the reader for the resource.
- * @throws IOException thrown if the resource could not be read.
- */
- public Reader getResourceAsReader(String path, int maxLen) throws IOException {
- return new InputStreamReader(getResourceAsStream(path, maxLen));
- }
-
- /**
- * Copies an char input stream into an char output stream.
- *
- * @param reader reader to copy from.
- * @param writer writer to copy to.
- * @throws IOException thrown if the copy failed.
- */
- public void copyCharStream(Reader reader, Writer writer) throws IOException {
- char[] buffer = new char[4096];
- int read;
- while ((read = reader.read(buffer)) > -1) {
- writer.write(buffer, 0, read);
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/oozie/blob/a54f7c20/minitest/src/test/resources/fs-decision.xml
----------------------------------------------------------------------
diff --git a/minitest/src/test/resources/fs-decision.xml b/minitest/src/test/resources/fs-decision.xml
new file mode 100644
index 0000000..72ac07a
--- /dev/null
+++ b/minitest/src/test/resources/fs-decision.xml
@@ -0,0 +1,51 @@
+<!--
+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.
+-->
+<workflow-app xmlns="uri:oozie:workflow:0.1" name="fs-decision">
+ <start to="fs1"/>
+ <action name="fs1">
+ <fs>
+ <mkdir path="${nameNodeBasePath}/p1"/>
+ </fs>
+ <ok to="fs2"/>
+ <error to="k"/>
+ </action>
+ <action name="fs2">
+ <fs>
+ <move source="${nameNodeBasePath}/p1" target="${base}/p2"/>
+ </fs>
+ <ok to="dec3"/>
+ <error to="k"/>
+ </action>
+ <decision name="dec3">
+ <switch>
+ <case to="fs4">${fs:exists(concat(nameNodeBasePath, '/p2'))}</case>
+ <default to="k"/>
+ </switch>
+ </decision>
+ <action name="fs4">
+ <fs>
+ <move source="${nameNodeBasePath}/p2" target="${base}/p3"/>
+ </fs>
+ <ok to="end"/>
+ <error to="k"/>
+ </action>
+ <kill name="k">
+ <message>kill</message>
+ </kill>
+ <end name="end"/>
+</workflow-app>
http://git-wip-us.apache.org/repos/asf/oozie/blob/a54f7c20/minitest/src/test/resources/hsqldb-oozie-site.xml
----------------------------------------------------------------------
diff --git a/minitest/src/test/resources/hsqldb-oozie-site.xml b/minitest/src/test/resources/hsqldb-oozie-site.xml
index fa5fe9c..2cdce74 100644
--- a/minitest/src/test/resources/hsqldb-oozie-site.xml
+++ b/minitest/src/test/resources/hsqldb-oozie-site.xml
@@ -19,8 +19,12 @@
-->
<configuration>
<property>
+ <name>oozie.service.JPAService.connection.data.source</name>
+ <value>org.apache.oozie.util.db.BasicDataSourceWrapper</value>
+ </property>
+ <property>
<name>oozie.service.JPAService.jdbc.driver</name>
- <value>org.hsqldb.jdbcDriver</value>
+ <value>org.apache.oozie.util.db.FailingHSQLDBDriverWrapper</value>
</property>
<property>
<name>oozie.service.JPAService.jdbc.url</name>
http://git-wip-us.apache.org/repos/asf/oozie/blob/a54f7c20/minitest/src/test/resources/oozie-log4j.properties
----------------------------------------------------------------------
diff --git a/minitest/src/test/resources/oozie-log4j.properties b/minitest/src/test/resources/oozie-log4j.properties
index c142d72..2503851 100644
--- a/minitest/src/test/resources/oozie-log4j.properties
+++ b/minitest/src/test/resources/oozie-log4j.properties
@@ -34,3 +34,4 @@ log4j.logger.org.apache.openjpa=INFO, test
log4j.logger.org.apache.hadoop=INFO, test
log4j.logger.org.mortbay=INFO, test
log4j.logger.org.hsqldb=INFO, test
+log4j.logger.org.apache.oozie.util.db=TRACE, test
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/oozie/blob/a54f7c20/minitest/src/test/resources/parallel-fs-and-shell.xml
----------------------------------------------------------------------
diff --git a/minitest/src/test/resources/parallel-fs-and-shell.xml b/minitest/src/test/resources/parallel-fs-and-shell.xml
new file mode 100644
index 0000000..7118fad
--- /dev/null
+++ b/minitest/src/test/resources/parallel-fs-and-shell.xml
@@ -0,0 +1,73 @@
+<!--
+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.
+-->
+<workflow-app name="parallel-fs-and-shell" xmlns="uri:oozie:workflow:0.5">
+ <start to="decision"/>
+ <kill name="kill">
+ <message>Action failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>
+ </kill>
+ <action name="fs1">
+ <fs>
+ <delete path='${nameNodeBasePath}/user/${wf:user()}/fs1'/>
+ <mkdir path='${nameNodeBasePath}/user/${wf:user()}/fs1'/>
+ <touchz path='${nameNodeBasePath}/user/${wf:user()}/fs1/fs1.out'/>
+ </fs>
+ <ok to="shell1"/>
+ <error to="kill"/>
+ </action>
+ <action name="fs2">
+ <fs>
+ <delete path='${nameNodeBasePath}/user/${wf:user()}/fs2'/>
+ <mkdir path='${nameNodeBasePath}/user/${wf:user()}/fs2'/>
+ <touchz path='${nameNodeBasePath}/user/${wf:user()}/fs2/fs2.out'/>
+ </fs>
+ <ok to="shell2"/>
+ <error to="kill"/>
+ </action>
+ <decision name="decision">
+ <switch>
+ <case to="fs1">
+ ${ choosePath1 }
+ </case>
+ <case to="fs2">
+ ${ !choosePath1 }
+ </case>
+ <default to="End"/>
+ </switch>
+ </decision>
+ <action name="shell1">
+ <shell xmlns="uri:oozie:shell-action:0.1">
+ <job-tracker>${jobTracker}</job-tracker>
+ <name-node>${nameNode}</name-node>
+ <exec>ls</exec>
+ <capture-output/>
+ </shell>
+ <ok to="End"/>
+ <error to="kill"/>
+ </action>
+ <action name="shell2">
+ <shell xmlns="uri:oozie:shell-action:0.1">
+ <job-tracker>${jobTracker}</job-tracker>
+ <name-node>${nameNode}</name-node>
+ <exec>ls</exec>
+ <capture-output/>
+ </shell>
+ <ok to="End"/>
+ <error to="kill"/>
+ </action>
+ <end name="End"/>
+</workflow-app>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/oozie/blob/a54f7c20/minitest/src/test/resources/wf-test.xml
----------------------------------------------------------------------
diff --git a/minitest/src/test/resources/wf-test.xml b/minitest/src/test/resources/wf-test.xml
deleted file mode 100644
index 20c4946..0000000
--- a/minitest/src/test/resources/wf-test.xml
+++ /dev/null
@@ -1,51 +0,0 @@
-<!--
-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.
--->
-<workflow-app xmlns="uri:oozie:workflow:0.1" name="recovery-wf">
- <start to="fs1"/>
- <action name="fs1">
- <fs>
- <mkdir path="${nnbase}/p1"/>
- </fs>
- <ok to="fs2"/>
- <error to="k"/>
- </action>
- <action name="fs2">
- <fs>
- <move source="${nnbase}/p1" target="${base}/p2"/>
- </fs>
- <ok to="dec3"/>
- <error to="k"/>
- </action>
- <decision name="dec3">
- <switch>
- <case to="fs4">${fs:exists(concat(nnbase, '/p2'))}</case>
- <default to="k"/>
- </switch>
- </decision>
- <action name="fs4">
- <fs>
- <move source="${nnbase}/p2" target="${base}/p3"/>
- </fs>
- <ok to="end"/>
- <error to="k"/>
- </action>
- <kill name="k">
- <message>kill</message>
- </kill>
- <end name="end"/>
-</workflow-app>
http://git-wip-us.apache.org/repos/asf/oozie/blob/a54f7c20/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 16c5137..0d1105b 100644
--- a/pom.xml
+++ b/pom.xml
@@ -1282,7 +1282,7 @@
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
- <version>5.1.6</version>
+ <version>5.1.34</version>
</dependency>
<dependency>
http://git-wip-us.apache.org/repos/asf/oozie/blob/a54f7c20/release-log.txt
----------------------------------------------------------------------
diff --git a/release-log.txt b/release-log.txt
index cf5956a..7443f2c 100644
--- a/release-log.txt
+++ b/release-log.txt
@@ -1,5 +1,6 @@
-- Oozie 5.0.0 release (trunk - unreleased)
+OOZIE-2854 Oozie should handle transient database problems (andras.piros via gezapeti)
OOZIE-2371 Add docs for state transitions for WF Action states (daniel.becker via gezapeti)
OOZIE-2911 Re-add test testWfActionKillChildJob and adapt it to OYA (gezapeti)
OOZIE-2918 Delete LauncherMapper and its test (asasvari via pbacsko)
[2/2] oozie git commit: OOZIE-2854 Oozie should handle transient
database problems (andras.piros via gezapeti)
Posted by ge...@apache.org.
OOZIE-2854 Oozie should handle transient database problems (andras.piros via gezapeti)
Project: http://git-wip-us.apache.org/repos/asf/oozie/repo
Commit: http://git-wip-us.apache.org/repos/asf/oozie/commit/a54f7c20
Tree: http://git-wip-us.apache.org/repos/asf/oozie/tree/a54f7c20
Diff: http://git-wip-us.apache.org/repos/asf/oozie/diff/a54f7c20
Branch: refs/heads/master
Commit: a54f7c20d7c5d1d72f195d6d92001edd147bc304
Parents: a436d98
Author: Gezapeti Cseh <ge...@gmail.com>
Authored: Wed Jul 12 09:16:10 2017 +0200
Committer: Gezapeti Cseh <ge...@gmail.com>
Committed: Wed Jul 12 09:16:25 2017 +0200
----------------------------------------------------------------------
core/pom.xml | 6 +
.../oozie/command/SkipCommitFaultInjection.java | 41 ++
.../executor/jpa/JsonBeanPersisterExecutor.java | 42 +++
.../oozie/executor/jpa/QueryExecutor.java | 25 +-
.../org/apache/oozie/service/JPAService.java | 371 +++++++++++++------
.../org/apache/oozie/sla/SLASummaryBean.java | 5 +-
.../org/apache/oozie/store/WorkflowStore.java | 1 -
.../oozie/util/db/BasicDataSourceWrapper.java | 123 ++++++
.../oozie/util/db/DatabaseRetryPredicate.java | 51 +++
.../oozie/util/db/FailingConnectionWrapper.java | 371 +++++++++++++++++++
.../util/db/FailingHSQLDBDriverWrapper.java | 37 ++
.../util/db/FailingMySQLDriverWrapper.java | 51 +++
.../oozie/util/db/OperationRetryHandler.java | 130 +++++++
...ceExceptionSubclassFilterRetryPredicate.java | 73 ++++
.../apache/oozie/util/db/RetryAttemptState.java | 84 +++++
.../oozie/util/db/RuntimeExceptionInjector.java | 77 ++++
.../src/main/resources/META-INF/persistence.xml | 12 +-
core/src/main/resources/oozie-default.xml | 33 +-
core/src/main/resources/oozie-log4j.properties | 1 +
.../oozie/command/SkipCommitFaultInjection.java | 41 --
.../java/org/apache/oozie/test/XTestCase.java | 106 +++---
.../util/db/TestOozieDmlStatementPredicate.java | 63 ++++
.../util/db/TestOperationRetryHandler.java | 218 +++++++++++
...ceExceptionSubclassFilterRetryPredicate.java | 98 +++++
.../oozie/util/db/TestRetryAttemptState.java | 132 +++++++
docs/src/site/twiki/AG_Install.twiki | 31 ++
minitest/pom.xml | 4 +-
.../test/TestParallelJPAOperationRetries.java | 144 +++++++
.../org/apache/oozie/test/TestWorkflow.java | 215 +++++++++++
.../apache/oozie/test/TestWorkflowRetries.java | 34 ++
.../org/apache/oozie/test/WorkflowTest.java | 188 ----------
minitest/src/test/resources/fs-decision.xml | 51 +++
.../src/test/resources/hsqldb-oozie-site.xml | 6 +-
.../src/test/resources/oozie-log4j.properties | 1 +
.../test/resources/parallel-fs-and-shell.xml | 73 ++++
minitest/src/test/resources/wf-test.xml | 51 ---
pom.xml | 2 +-
release-log.txt | 1 +
38 files changed, 2492 insertions(+), 501 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/oozie/blob/a54f7c20/core/pom.xml
----------------------------------------------------------------------
diff --git a/core/pom.xml b/core/pom.xml
index acddf34..7275775 100644
--- a/core/pom.xml
+++ b/core/pom.xml
@@ -280,6 +280,12 @@
</dependency>
<dependency>
+ <groupId>mysql</groupId>
+ <artifactId>mysql-connector-java</artifactId>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
<groupId>commons-logging</groupId>
<artifactId>commons-logging</artifactId>
<scope>compile</scope>
http://git-wip-us.apache.org/repos/asf/oozie/blob/a54f7c20/core/src/main/java/org/apache/oozie/command/SkipCommitFaultInjection.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/oozie/command/SkipCommitFaultInjection.java b/core/src/main/java/org/apache/oozie/command/SkipCommitFaultInjection.java
new file mode 100644
index 0000000..158fbfa
--- /dev/null
+++ b/core/src/main/java/org/apache/oozie/command/SkipCommitFaultInjection.java
@@ -0,0 +1,41 @@
+/**
+ * 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.oozie.command;
+
+import org.apache.oozie.FaultInjection;
+
+public class SkipCommitFaultInjection extends FaultInjection {
+
+ public static final String ACTION_FAILOVER_FAULT_INJECTION = "oozie.fault.injection.action.failover";
+
+ private static boolean ACTIVE = false;
+
+ public boolean activate() {
+ ACTIVE = Boolean.parseBoolean(System.getProperty(ACTION_FAILOVER_FAULT_INJECTION, "false"));
+ return ACTIVE;
+ }
+
+ public void deactivate() {
+ ACTIVE = false;
+ }
+
+ public boolean isActive() {
+ return ACTIVE;
+ }
+}
http://git-wip-us.apache.org/repos/asf/oozie/blob/a54f7c20/core/src/main/java/org/apache/oozie/executor/jpa/JsonBeanPersisterExecutor.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/oozie/executor/jpa/JsonBeanPersisterExecutor.java b/core/src/main/java/org/apache/oozie/executor/jpa/JsonBeanPersisterExecutor.java
new file mode 100644
index 0000000..217e442
--- /dev/null
+++ b/core/src/main/java/org/apache/oozie/executor/jpa/JsonBeanPersisterExecutor.java
@@ -0,0 +1,42 @@
+/**
+ * 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.oozie.executor.jpa;
+
+import javax.persistence.EntityManager;
+
+import org.apache.oozie.client.rest.JsonBean;
+
+public class JsonBeanPersisterExecutor implements JPAExecutor<Void> {
+ private final JsonBean bean;
+
+ public JsonBeanPersisterExecutor(JsonBean bean) {
+ this.bean = bean;
+ }
+
+ @Override
+ public String getName() {
+ return "JsonBeanPersisterExecutor";
+ }
+
+ @Override
+ public Void execute(EntityManager em) throws JPAExecutorException {
+ em.persist(bean);
+ return null;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/oozie/blob/a54f7c20/core/src/main/java/org/apache/oozie/executor/jpa/QueryExecutor.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/oozie/executor/jpa/QueryExecutor.java b/core/src/main/java/org/apache/oozie/executor/jpa/QueryExecutor.java
index 8d94c23..dfafea0 100644
--- a/core/src/main/java/org/apache/oozie/executor/jpa/QueryExecutor.java
+++ b/core/src/main/java/org/apache/oozie/executor/jpa/QueryExecutor.java
@@ -21,47 +21,26 @@ package org.apache.oozie.executor.jpa;
import java.util.List;
import javax.persistence.EntityManager;
-import javax.persistence.PersistenceException;
import javax.persistence.Query;
-import org.apache.oozie.ErrorCode;
import org.apache.oozie.client.rest.JsonBean;
import org.apache.oozie.service.JPAService;
import org.apache.oozie.service.Services;
-import org.apache.oozie.util.XLog;
/**
* Base Class of Query Executor
*/
public abstract class QueryExecutor<T, E extends Enum<E>> {
- private static XLog LOG;
protected QueryExecutor() {
}
public abstract int executeUpdate(E namedQuery, T jobBean) throws JPAExecutorException;
- public void insert(JsonBean bean) throws JPAExecutorException {
+ public void insert(final JsonBean bean) throws JPAExecutorException {
if (bean != null) {
JPAService jpaService = Services.get().get(JPAService.class);
- EntityManager em = jpaService.getEntityManager();
- try {
- em.getTransaction().begin();
- em.persist(bean);
- em.getTransaction().commit();
- }
- catch (PersistenceException e) {
- throw new JPAExecutorException(ErrorCode.E0603, e);
- }
- finally {
- if (em.getTransaction().isActive()) {
- LOG.warn("insert ended with an active transaction, rolling back");
- em.getTransaction().rollback();
- }
- if (em.isOpen()) {
- em.close();
- }
- }
+ jpaService.execute(new JsonBeanPersisterExecutor(bean));
}
}
http://git-wip-us.apache.org/repos/asf/oozie/blob/a54f7c20/core/src/main/java/org/apache/oozie/service/JPAService.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/oozie/service/JPAService.java b/core/src/main/java/org/apache/oozie/service/JPAService.java
index 028381d..93fe9da 100644
--- a/core/src/main/java/org/apache/oozie/service/JPAService.java
+++ b/core/src/main/java/org/apache/oozie/service/JPAService.java
@@ -23,14 +23,16 @@ import java.text.MessageFormat;
import java.util.Collection;
import java.util.List;
import java.util.Properties;
+import java.util.concurrent.Callable;
import javax.persistence.EntityManager;
import javax.persistence.EntityManagerFactory;
+import javax.persistence.EntityTransaction;
import javax.persistence.NoResultException;
import javax.persistence.Persistence;
-import javax.persistence.PersistenceException;
import javax.persistence.Query;
+import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.dbcp.BasicDataSource;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
@@ -45,6 +47,7 @@ import org.apache.oozie.WorkflowActionBean;
import org.apache.oozie.WorkflowJobBean;
import org.apache.oozie.client.rest.JsonBean;
import org.apache.oozie.client.rest.JsonSLAEvent;
+import org.apache.oozie.command.SkipCommitFaultInjection;
import org.apache.oozie.compression.CodecFactory;
import org.apache.oozie.executor.jpa.JPAExecutor;
import org.apache.oozie.executor.jpa.JPAExecutorException;
@@ -54,7 +57,10 @@ import org.apache.oozie.util.IOUtils;
import org.apache.oozie.util.Instrumentable;
import org.apache.oozie.util.Instrumentation;
import org.apache.oozie.util.XLog;
+import org.apache.oozie.util.db.OperationRetryHandler;
+import org.apache.oozie.util.db.PersistenceExceptionSubclassFilterRetryPredicate;
import org.apache.openjpa.lib.jdbc.DecoratingDataSource;
+import org.apache.openjpa.persistence.InvalidStateException;
import org.apache.openjpa.persistence.OpenJPAEntityManagerFactorySPI;
/**
@@ -64,6 +70,10 @@ import org.apache.openjpa.persistence.OpenJPAEntityManagerFactorySPI;
public class JPAService implements Service, Instrumentable {
private static final String INSTRUMENTATION_GROUP_JPA = "jpa";
+ public static final long DEFAULT_INITIAL_WAIT_TIME = 100;
+ public static final long DEFAULT_MAX_WAIT_TIME = 30_000;
+ public static final int DEFAULT_MAX_RETRY_COUNT = 1;
+
public static final String CONF_DB_SCHEMA = "oozie.db.schema.name";
public static final String CONF_PREFIX = Service.CONF_PREFIX + "JPAService.";
@@ -79,13 +89,16 @@ public class JPAService implements Service, Instrumentable {
public static final String CONF_VALIDATE_DB_CONN_EVICTION_INTERVAL = CONF_PREFIX + "validate.db.connection.eviction.interval";
public static final String CONF_VALIDATE_DB_CONN_EVICTION_NUM = CONF_PREFIX + "validate.db.connection.eviction.num";
public static final String CONF_OPENJPA_BROKER_IMPL = CONF_PREFIX + "openjpa.BrokerImpl";
-
-
+ public static final String INITIAL_WAIT_TIME = CONF_PREFIX + "retry.initial-wait-time.ms";
+ public static final String MAX_WAIT_TIME = CONF_PREFIX + "maximum-wait-time.ms";
+ public static final String MAX_RETRY_COUNT = CONF_PREFIX + "retry.max-retries";
+ public static final String SKIP_COMMIT_FAULT_INJECTION_CLASS = SkipCommitFaultInjection.class.getName();
private EntityManagerFactory factory;
private Instrumentation instr;
private static XLog LOG;
+ private OperationRetryHandler retryHandler;
/**
* Return the public interface of the service.
@@ -97,7 +110,7 @@ public class JPAService implements Service, Instrumentable {
}
@Override
- public void instrument(Instrumentation instr) {
+ public void instrument(final Instrumentation instr) {
this.instr = instr;
final BasicDataSource dataSource = getBasicDataSource();
@@ -121,10 +134,10 @@ public class JPAService implements Service, Instrumentable {
// Get the BasicDataSource object; it could be wrapped in a DecoratingDataSource
// It might also not be a BasicDataSource if the user configured something different
BasicDataSource basicDataSource = null;
- OpenJPAEntityManagerFactorySPI spi = (OpenJPAEntityManagerFactorySPI) factory;
- Object connectionFactory = spi.getConfiguration().getConnectionFactory();
+ final OpenJPAEntityManagerFactorySPI spi = (OpenJPAEntityManagerFactorySPI) factory;
+ final Object connectionFactory = spi.getConfiguration().getConnectionFactory();
if (connectionFactory instanceof DecoratingDataSource) {
- DecoratingDataSource decoratingDataSource = (DecoratingDataSource) connectionFactory;
+ final DecoratingDataSource decoratingDataSource = (DecoratingDataSource) connectionFactory;
basicDataSource = (BasicDataSource) decoratingDataSource.getInnermostDelegate();
} else if (connectionFactory instanceof BasicDataSource) {
basicDataSource = (BasicDataSource) connectionFactory;
@@ -137,22 +150,22 @@ public class JPAService implements Service, Instrumentable {
*
* @param services services instance.
*/
- public void init(Services services) throws ServiceException {
+ public void init(final Services services) throws ServiceException {
LOG = XLog.getLog(JPAService.class);
- Configuration conf = services.getConf();
- String dbSchema = ConfigurationService.get(conf, CONF_DB_SCHEMA);
+ final Configuration conf = services.getConf();
+ final String dbSchema = ConfigurationService.get(conf, CONF_DB_SCHEMA);
String url = ConfigurationService.get(conf, CONF_URL);
- String driver = ConfigurationService.get(conf, CONF_DRIVER);
- String user = ConfigurationService.get(conf, CONF_USERNAME);
- String password = ConfigurationService.getPassword(conf, CONF_PASSWORD).trim();
- String maxConn = ConfigurationService.get(conf, CONF_MAX_ACTIVE_CONN).trim();
- String dataSource = ConfigurationService.get(conf, CONF_CONN_DATA_SOURCE);
- String connPropsConfig = ConfigurationService.get(conf, CONF_CONN_PROPERTIES);
- String brokerImplConfig = ConfigurationService.get(conf, CONF_OPENJPA_BROKER_IMPL);
- boolean autoSchemaCreation = ConfigurationService.getBoolean(conf, CONF_CREATE_DB_SCHEMA);
- boolean validateDbConn = ConfigurationService.getBoolean(conf, CONF_VALIDATE_DB_CONN);
- String evictionInterval = ConfigurationService.get(conf, CONF_VALIDATE_DB_CONN_EVICTION_INTERVAL).trim();
- String evictionNum = ConfigurationService.get(conf, CONF_VALIDATE_DB_CONN_EVICTION_NUM).trim();
+ final String driver = ConfigurationService.get(conf, CONF_DRIVER);
+ final String user = ConfigurationService.get(conf, CONF_USERNAME);
+ final String password = ConfigurationService.getPassword(conf, CONF_PASSWORD).trim();
+ final String maxConn = ConfigurationService.get(conf, CONF_MAX_ACTIVE_CONN).trim();
+ final String dataSource = ConfigurationService.get(conf, CONF_CONN_DATA_SOURCE);
+ final String connPropsConfig = ConfigurationService.get(conf, CONF_CONN_PROPERTIES);
+ final String brokerImplConfig = ConfigurationService.get(conf, CONF_OPENJPA_BROKER_IMPL);
+ final boolean autoSchemaCreation = ConfigurationService.getBoolean(conf, CONF_CREATE_DB_SCHEMA);
+ final boolean validateDbConn = ConfigurationService.getBoolean(conf, CONF_VALIDATE_DB_CONN);
+ final String evictionInterval = ConfigurationService.get(conf, CONF_VALIDATE_DB_CONN_EVICTION_INTERVAL).trim();
+ final String evictionNum = ConfigurationService.get(conf, CONF_VALIDATE_DB_CONN_EVICTION_NUM).trim();
if (!url.startsWith("jdbc:")) {
throw new ServiceException(ErrorCode.E0608, url, "invalid JDBC URL, must start with 'jdbc:'");
@@ -163,14 +176,14 @@ public class JPAService implements Service, Instrumentable {
}
dbType = dbType.substring(0, dbType.indexOf(":"));
- String persistentUnit = "oozie-" + dbType;
+ final String persistentUnit = "oozie-" + dbType;
// Checking existince of ORM file for DB type
- String ormFile = "META-INF/" + persistentUnit + "-orm.xml";
+ final String ormFile = "META-INF/" + persistentUnit + "-orm.xml";
try {
IOUtils.getResourceAsStream(ormFile, -1);
}
- catch (IOException ex) {
+ catch (final IOException ex) {
throw new ServiceException(ErrorCode.E0609, dbType, ormFile);
}
@@ -182,7 +195,7 @@ public class JPAService implements Service, Instrumentable {
String connProps = "DriverClassName={0},Url={1},Username={2},Password={3},MaxActive={4}";
connProps = MessageFormat.format(connProps, driver, url, user, password, maxConn);
- Properties props = new Properties();
+ final Properties props = new Properties();
if (autoSchemaCreation) {
connProps += ",TestOnBorrow=false,TestOnReturn=false,TestWhileIdle=false";
props.setProperty("openjpa.jdbc.SynchronizeMappings", "buildSchema(ForeignKeys=true)");
@@ -190,8 +203,8 @@ public class JPAService implements Service, Instrumentable {
else if (validateDbConn) {
// validation can be done only if the schema already exist, else a
// connection cannot be obtained to create the schema.
- String interval = "timeBetweenEvictionRunsMillis=" + evictionInterval;
- String num = "numTestsPerEvictionRun=" + evictionNum;
+ final String interval = "timeBetweenEvictionRunsMillis=" + evictionInterval;
+ final String num = "numTestsPerEvictionRun=" + evictionNum;
connProps += ",TestOnBorrow=true,TestOnReturn=true,TestWhileIdle=true," + interval + "," + num;
connProps += ",ValidationQuery=select count(*) from VALIDATE_CONN";
connProps = MessageFormat.format(connProps, dbSchema);
@@ -210,36 +223,76 @@ public class JPAService implements Service, Instrumentable {
LOG.info("Setting openjpa.BrokerImpl to {0}", brokerImplConfig);
}
+ initRetryHandler();
+
factory = Persistence.createEntityManagerFactory(persistentUnit, props);
- EntityManager entityManager = getEntityManager();
- entityManager.find(WorkflowActionBean.class, 1);
- entityManager.find(WorkflowJobBean.class, 1);
- entityManager.find(CoordinatorActionBean.class, 1);
- entityManager.find(CoordinatorJobBean.class, 1);
- entityManager.find(SLAEventBean.class, 1);
- entityManager.find(JsonSLAEvent.class, 1);
- entityManager.find(BundleJobBean.class, 1);
- entityManager.find(BundleActionBean.class, 1);
- entityManager.find(SLARegistrationBean.class, 1);
- entityManager.find(SLASummaryBean.class, 1);
+ final EntityManager entityManager = getEntityManager();
+ findRetrying(entityManager, WorkflowActionBean.class, 1);
+ findRetrying(entityManager, WorkflowJobBean.class, 1);
+ findRetrying(entityManager, CoordinatorActionBean.class, 1);
+ findRetrying(entityManager, CoordinatorJobBean.class, 1);
+ findRetrying(entityManager, SLAEventBean.class, 1);
+ findRetrying(entityManager, JsonSLAEvent.class, 1);
+ findRetrying(entityManager, BundleActionBean.class, 1);
+ findRetrying(entityManager, BundleJobBean.class, 1);
+ findRetrying(entityManager, SLARegistrationBean.class, 1);
+ findRetrying(entityManager, SLASummaryBean.class, 1);
LOG.info(XLog.STD, "All entities initialized");
// need to use a pseudo no-op transaction so all entities, datasource
// and connection pool are initialized one time only
entityManager.getTransaction().begin();
- OpenJPAEntityManagerFactorySPI spi = (OpenJPAEntityManagerFactorySPI) factory;
+ final OpenJPAEntityManagerFactorySPI spi = (OpenJPAEntityManagerFactorySPI) factory;
// Mask the password with '***'
- String logMsg = spi.getConfiguration().getConnectionProperties().replaceAll("Password=.*?,", "Password=***,");
+ final String logMsg = spi.getConfiguration().getConnectionProperties().replaceAll("Password=.*?,", "Password=***,");
LOG.info("JPA configuration: {0}", logMsg);
entityManager.getTransaction().commit();
entityManager.close();
try {
CodecFactory.initialize(conf);
}
- catch (Exception ex) {
+ catch (final Exception ex) {
throw new ServiceException(ErrorCode.E0100, getClass().getName(), ex);
}
+
+ }
+
+ private void initRetryHandler() {
+ final long initialWaitTime = ConfigurationService.getInt(INITIAL_WAIT_TIME, (int) DEFAULT_INITIAL_WAIT_TIME);
+ final long maxWaitTime = ConfigurationService.getInt(MAX_WAIT_TIME, (int) DEFAULT_MAX_WAIT_TIME);
+ final int maxRetryCount = ConfigurationService.getInt(MAX_RETRY_COUNT, DEFAULT_MAX_RETRY_COUNT);
+
+ LOG.info(XLog.STD, "Failing database operations will be retried {0} times, with an initial sleep time of {1} ms,"
+ + "max sleep time {2} ms", maxRetryCount, initialWaitTime, maxWaitTime);
+ retryHandler = new OperationRetryHandler(maxRetryCount,
+ initialWaitTime,
+ maxWaitTime,
+ new PersistenceExceptionSubclassFilterRetryPredicate());
+ }
+
+ private void findRetrying(final EntityManager entityManager, final Class entityClass, final int primaryKey)
+ throws ServiceException {
+ try {
+ retryHandler.executeWithRetry(new Callable<Void>() {
+ @Override
+ public Void call() throws Exception {
+ if (!entityManager.getTransaction().isActive()) {
+ entityManager.getTransaction().begin();
+ }
+
+ entityManager.find(entityClass, primaryKey);
+
+ if (entityManager.getTransaction().isActive()) {
+ entityManager.getTransaction().commit();
+ }
+ return null;
+ }
+ });
+ }
+ catch (final Exception e) {
+ throw new ServiceException(ErrorCode.E0603, e);
+ }
}
/**
@@ -247,7 +300,12 @@ public class JPAService implements Service, Instrumentable {
*/
public void destroy() {
if (factory != null && factory.isOpen()) {
- factory.close();
+ try {
+ factory.close();
+ }
+ catch (final InvalidStateException ise) {
+ LOG.warn("Cannot close EntityManagerFactory. [ise.message={0}]", ise.getMessage());
+ }
}
}
@@ -258,28 +316,33 @@ public class JPAService implements Service, Instrumentable {
* @return return value of the JPAExecutor.
* @throws JPAExecutorException thrown if an jpa executor failed
*/
- public <T> T execute(JPAExecutor<T> executor) throws JPAExecutorException {
- EntityManager em = getEntityManager();
- Instrumentation.Cron cron = new Instrumentation.Cron();
+ public <T> T execute(final JPAExecutor<T> executor) throws JPAExecutorException {
+ final EntityManager em = getEntityManager();
+ final Instrumentation.Cron cron = new Instrumentation.Cron();
try {
LOG.trace("Executing JPAExecutor [{0}]", executor.getName());
if (instr != null) {
instr.incr(INSTRUMENTATION_GROUP_JPA, executor.getName(), 1);
}
cron.start();
- em.getTransaction().begin();
- T t = executor.execute(em);
- if (em.getTransaction().isActive()) {
- if (FaultInjection.isActive("org.apache.oozie.command.SkipCommitFaultInjection")) {
- throw new RuntimeException("Skipping Commit for Failover Testing");
- }
- em.getTransaction().commit();
- }
- return t;
+ return retryHandler.executeWithRetry(new Callable<T>() {
+ @Override
+ public T call() throws Exception {
+ if (!em.getTransaction().isActive()) {
+ em.getTransaction().begin();
+ }
+
+ final T t = executor.execute(em);
+
+ checkAndCommit(em.getTransaction());
+
+ return t;
+ }
+ });
}
- catch (PersistenceException e) {
- throw new JPAExecutorException(ErrorCode.E0603, e);
+ catch (final Exception e) {
+ throw getTargetException(e);
}
finally {
cron.stop();
@@ -292,7 +355,7 @@ public class JPAService implements Service, Instrumentable {
em.getTransaction().rollback();
}
}
- catch (Exception ex) {
+ catch (final Exception ex) {
LOG.warn("Could not check/rollback transaction after JPAExecutor [{0}], {1}", executor.getName(), ex
.getMessage(), ex);
}
@@ -304,13 +367,23 @@ public class JPAService implements Service, Instrumentable {
LOG.warn("JPAExecutor [{0}] closed the EntityManager, it should not!", executor.getName());
}
}
- catch (Exception ex) {
+ catch (final Exception ex) {
LOG.warn("Could not close EntityManager after JPAExecutor [{0}], {1}", executor.getName(), ex
.getMessage(), ex);
}
}
}
+ private void checkAndCommit(final EntityTransaction tx) {
+ if (tx.isActive()) {
+ if (FaultInjection.isActive(SKIP_COMMIT_FAULT_INJECTION_CLASS)) {
+ throw new RuntimeException("Skipping Commit for Failover Testing");
+ }
+
+ tx.commit();
+ }
+ }
+
/**
* Execute an UPDATE query
* @param namedQueryName the name of query to be executed
@@ -319,8 +392,8 @@ public class JPAService implements Service, Instrumentable {
* @return Integer that query returns, which corresponds to the number of rows updated
* @throws JPAExecutorException
*/
- public int executeUpdate(String namedQueryName, Query query, EntityManager em) throws JPAExecutorException {
- Instrumentation.Cron cron = new Instrumentation.Cron();
+ public int executeUpdate(final String namedQueryName, final Query query, final EntityManager em) throws JPAExecutorException {
+ final Instrumentation.Cron cron = new Instrumentation.Cron();
try {
LOG.trace("Executing Update/Delete Query [{0}]", namedQueryName);
@@ -328,18 +401,23 @@ public class JPAService implements Service, Instrumentable {
instr.incr(INSTRUMENTATION_GROUP_JPA, namedQueryName, 1);
}
cron.start();
- em.getTransaction().begin();
- int ret = query.executeUpdate();
- if (em.getTransaction().isActive()) {
- if (FaultInjection.isActive("org.apache.oozie.command.SkipCommitFaultInjection")) {
- throw new RuntimeException("Skipping Commit for Failover Testing");
+
+ return retryHandler.executeWithRetry(new Callable<Integer>() {
+ @Override
+ public Integer call() throws Exception {
+ if (!em.getTransaction().isActive()) {
+ em.getTransaction().begin();
+ }
+ final int ret = query.executeUpdate();
+
+ checkAndCommit(em.getTransaction());
+
+ return ret;
}
- em.getTransaction().commit();
- }
- return ret;
+ });
}
- catch (PersistenceException e) {
- throw new JPAExecutorException(ErrorCode.E0603, e);
+ catch (final Exception e) {
+ throw getTargetException(e);
}
finally {
processFinally(em, cron, namedQueryName, true);
@@ -350,7 +428,7 @@ public class JPAService implements Service, Instrumentable {
E namedQuery;
Query query;
- public QueryEntry(E namedQuery, Query query) {
+ public QueryEntry(final E namedQuery, final Query query) {
this.namedQuery = namedQuery;
this.query = query;
}
@@ -364,7 +442,10 @@ public class JPAService implements Service, Instrumentable {
}
}
- private void processFinally(EntityManager em, Instrumentation.Cron cron, String name, boolean checkActive) {
+ private void processFinally(final EntityManager em,
+ final Instrumentation.Cron cron,
+ final String name,
+ final boolean checkActive) {
cron.stop();
if (instr != null) {
instr.addCron(INSTRUMENTATION_GROUP_JPA, name, cron);
@@ -376,7 +457,7 @@ public class JPAService implements Service, Instrumentable {
em.getTransaction().rollback();
}
}
- catch (Exception ex) {
+ catch (final Exception ex) {
LOG.warn("Could not check/rollback transaction after [{0}], {1}", name,
ex.getMessage(), ex);
}
@@ -389,7 +470,7 @@ public class JPAService implements Service, Instrumentable {
LOG.warn("[{0}] closed the EntityManager, it should not!", name);
}
}
- catch (Exception ex) {
+ catch (final Exception ex) {
LOG.warn("Could not close EntityManager after [{0}], {1}", name, ex.getMessage(), ex);
}
}
@@ -402,41 +483,57 @@ public class JPAService implements Service, Instrumentable {
* @param em Entity Manager
* @throws JPAExecutorException
*/
- public void executeBatchInsertUpdateDelete(Collection<JsonBean> insertBeans, List<QueryEntry> updateQueryList,
- Collection<JsonBean> deleteBeans, EntityManager em) throws JPAExecutorException {
- Instrumentation.Cron cron = new Instrumentation.Cron();
+ public void executeBatchInsertUpdateDelete(final Collection<JsonBean> insertBeans, final List<QueryEntry> updateQueryList,
+ final Collection<JsonBean> deleteBeans, final EntityManager em) throws JPAExecutorException {
+ final Instrumentation.Cron cron = new Instrumentation.Cron();
try {
LOG.trace("Executing Queries in Batch");
cron.start();
- em.getTransaction().begin();
- if (updateQueryList != null && updateQueryList.size() > 0) {
- for (QueryEntry q : updateQueryList) {
- if (instr != null) {
- instr.incr(INSTRUMENTATION_GROUP_JPA, q.getQueryName().name(), 1);
+
+ retryHandler.executeWithRetry(new Callable<Void>() {
+ @Override
+ public Void call() throws Exception {
+ if (em.getTransaction().isActive()) {
+ try {
+ em.getTransaction().rollback();
+ }
+ catch (final Exception e) {
+ LOG.warn("Rollback failed - ignoring");
+ }
}
- q.getQuery().executeUpdate();
- }
- }
- if (insertBeans != null && insertBeans.size() > 0) {
- for (JsonBean bean : insertBeans) {
- em.persist(bean);
- }
- }
- if (deleteBeans != null && deleteBeans.size() > 0) {
- for (JsonBean bean : deleteBeans) {
- em.remove(em.merge(bean));
- }
- }
- if (em.getTransaction().isActive()) {
- if (FaultInjection.isActive("org.apache.oozie.command.SkipCommitFaultInjection")) {
- throw new RuntimeException("Skipping Commit for Failover Testing");
+
+ em.getTransaction().begin();
+
+ if (CollectionUtils.isNotEmpty(updateQueryList)) {
+ for (final QueryEntry q : updateQueryList) {
+ if (instr != null) {
+ instr.incr(INSTRUMENTATION_GROUP_JPA, q.getQueryName().name(), 1);
+ }
+ q.getQuery().executeUpdate();
+ }
+ }
+
+ if (CollectionUtils.isNotEmpty(insertBeans)) {
+ for (final JsonBean bean : insertBeans) {
+ em.persist(bean);
+ }
+ }
+
+ if (CollectionUtils.isNotEmpty(deleteBeans)) {
+ for (final JsonBean bean : deleteBeans) {
+ em.remove(em.merge(bean));
+ }
+ }
+
+ checkAndCommit(em.getTransaction());
+
+ return null;
}
- em.getTransaction().commit();
- }
+ });
}
- catch (PersistenceException e) {
- throw new JPAExecutorException(ErrorCode.E0603, e);
+ catch (final Exception e) {
+ throw getTargetException(e);
}
finally {
processFinally(em, cron, "batchqueryexecutor", true);
@@ -450,24 +547,33 @@ public class JPAService implements Service, Instrumentable {
* @param em Entity Manager
* @return object that matches the query
*/
- public Object executeGet(String namedQueryName, Query query, EntityManager em) {
- Instrumentation.Cron cron = new Instrumentation.Cron();
+ public Object executeGet(final String namedQueryName, final Query query, final EntityManager em) throws JPAExecutorException {
+ final Instrumentation.Cron cron = new Instrumentation.Cron();
try {
-
LOG.trace("Executing Select Query to Get a Single row [{0}]", namedQueryName);
if (instr != null) {
instr.incr(INSTRUMENTATION_GROUP_JPA, namedQueryName, 1);
}
cron.start();
- Object obj = null;
- try {
- obj = query.getSingleResult();
- }
- catch (NoResultException e) {
- // return null when no matched result
- }
- return obj;
+
+ return retryHandler.executeWithRetry(new Callable<Object>() {
+ @Override
+ public Object call() throws Exception {
+ Object obj = null;
+ try {
+ obj = query.getSingleResult();
+ }
+ catch (final NoResultException e) {
+ LOG.info("No results found");
+ // return null when no matched result
+ }
+ return obj;
+ }
+ });
+ }
+ catch (final Exception e) {
+ throw getTargetException(e);
}
finally {
processFinally(em, cron, namedQueryName, false);
@@ -481,8 +587,9 @@ public class JPAService implements Service, Instrumentable {
* @param em Entity Manager
* @return list containing results that match the query
*/
- public List<?> executeGetList(String namedQueryName, Query query, EntityManager em) {
- Instrumentation.Cron cron = new Instrumentation.Cron();
+ public List<?> executeGetList(final String namedQueryName, final Query query, final EntityManager em)
+ throws JPAExecutorException {
+ final Instrumentation.Cron cron = new Instrumentation.Cron();
try {
LOG.trace("Executing Select Query to Get Multiple Rows [{0}]", namedQueryName);
@@ -491,14 +598,24 @@ public class JPAService implements Service, Instrumentable {
}
cron.start();
- List<?> resultList = null;
- try {
- resultList = query.getResultList();
- }
- catch (NoResultException e) {
- // return null when no matched result
- }
- return resultList;
+
+ return retryHandler.executeWithRetry(new Callable<List<?>>() {
+ @Override
+ public List<?> call() throws Exception {
+ List<?> resultList = null;
+ try {
+ resultList = query.getResultList();
+ }
+ catch (final NoResultException e) {
+ LOG.info("No results found");
+ // return null when no matched result
+ }
+ return resultList;
+ }
+ });
+ }
+ catch (final Exception e) {
+ throw getTargetException(e);
}
finally {
processFinally(em, cron, namedQueryName, false);
@@ -514,4 +631,12 @@ public class JPAService implements Service, Instrumentable {
return factory.createEntityManager();
}
+ private JPAExecutorException getTargetException(final Exception e) {
+ if (e instanceof JPAExecutorException) {
+ return (JPAExecutorException) e;
+ }
+ else {
+ return new JPAExecutorException(ErrorCode.E0603, e.getMessage());
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/oozie/blob/a54f7c20/core/src/main/java/org/apache/oozie/sla/SLASummaryBean.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/oozie/sla/SLASummaryBean.java b/core/src/main/java/org/apache/oozie/sla/SLASummaryBean.java
index cfe1522..e9ea9ba 100644
--- a/core/src/main/java/org/apache/oozie/sla/SLASummaryBean.java
+++ b/core/src/main/java/org/apache/oozie/sla/SLASummaryBean.java
@@ -66,7 +66,10 @@ import org.json.simple.JSONObject;
@NamedQuery(name = "GET_SLA_SUMMARY_EVENTPROCESSED", query = "select w.eventProcessed from SLASummaryBean w where w.jobId = :id"),
- @NamedQuery(name = "GET_SLA_SUMMARY_EVENTPROCESSED_LAST_MODIFIED", query = "select w.eventProcessed, w.lastModifiedTS from SLASummaryBean w where w.jobId = :id")
+ @NamedQuery(name = "GET_SLA_SUMMARY_EVENTPROCESSED_LAST_MODIFIED",
+ query = "select w.eventProcessed, w.lastModifiedTS from SLASummaryBean w where w.jobId = :id"),
+
+ @NamedQuery(name = "GET_SLA_SUMMARY_ALL", query = "select OBJECT(w) from SLASummaryBean w")
})
http://git-wip-us.apache.org/repos/asf/oozie/blob/a54f7c20/core/src/main/java/org/apache/oozie/store/WorkflowStore.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/oozie/store/WorkflowStore.java b/core/src/main/java/org/apache/oozie/store/WorkflowStore.java
index c565e74..821abc5 100644
--- a/core/src/main/java/org/apache/oozie/store/WorkflowStore.java
+++ b/core/src/main/java/org/apache/oozie/store/WorkflowStore.java
@@ -22,7 +22,6 @@ import java.sql.Connection;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.util.ArrayList;
-import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
http://git-wip-us.apache.org/repos/asf/oozie/blob/a54f7c20/core/src/main/java/org/apache/oozie/util/db/BasicDataSourceWrapper.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/oozie/util/db/BasicDataSourceWrapper.java b/core/src/main/java/org/apache/oozie/util/db/BasicDataSourceWrapper.java
new file mode 100644
index 0000000..c972ea2
--- /dev/null
+++ b/core/src/main/java/org/apache/oozie/util/db/BasicDataSourceWrapper.java
@@ -0,0 +1,123 @@
+/**
+ * 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.oozie.util.db;
+
+import org.apache.commons.dbcp.BasicDataSource;
+import org.apache.commons.dbcp.ConnectionFactory;
+import org.apache.commons.dbcp.DriverConnectionFactory;
+import org.apache.commons.dbcp.SQLNestedException;
+
+import java.sql.Driver;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+
+public class BasicDataSourceWrapper extends BasicDataSource {
+
+ /**
+ * Fixing a bug within {@link BasicDataSource#createConnectionFactory()} for {@code driverClassName} to have real effect.
+ * <p/>
+ * Because we use currently Apache Commons DBCP 1.4.0 that has a bug not considering {@code driverClassName}, thus, we're unable
+ * to create a JDBC driver using a user-provided driver class name (we try to do that by setting explicitly a value for
+ * {@code openJpa.connectionProperties="DriverClassName=..."}), unless we perform the exact same fix that is applied by the DBCP
+ * patch.
+ * <p/>
+ * Note: when DBCP 1.4.1 will be released, and Oozie will update to 1.4.1, we can remove this class.
+ * <p/>
+ * Please see {@link https://issues.apache.org/jira/browse/DBCP-333 the DBCP bug} and
+ * {@link https://github.com/apache/commons-dbcp/blob/DBCP_1_4_x_BRANCH/src/java/org/apache/commons/dbcp/BasicDataSource.java
+ * #L1588-L1660 the fixed method}
+ * for details.
+ * <p/>
+ * Please also see how OpenJPA
+ * {@linkplain http://openjpa.apache.org/builds/2.2.1/apache-openjpa/docs/ref_guide_integration_dbcp.html is integrated}
+ * with DBCP.
+ */
+ protected ConnectionFactory createConnectionFactory() throws SQLException {
+ // Load the JDBC driver class
+ Class driverFromCCL = null;
+ if (driverClassName != null) {
+ try {
+ try {
+ if (driverClassLoader == null) {
+ driverFromCCL = Class.forName(driverClassName);
+ } else {
+ driverFromCCL = Class.forName(driverClassName, true, driverClassLoader);
+ }
+ } catch (ClassNotFoundException cnfe) {
+ driverFromCCL = Thread.currentThread(
+ ).getContextClassLoader().loadClass(
+ driverClassName);
+ }
+ } catch (Throwable t) {
+ String message = "Cannot load JDBC driver class '" +
+ driverClassName + "'";
+ logWriter.println(message);
+ t.printStackTrace(logWriter);
+ throw new SQLNestedException(message, t);
+ }
+ }
+
+ // Create a JDBC driver instance
+ Driver driver = null;
+ try {
+ if (driverFromCCL == null) {
+ driver = DriverManager.getDriver(url);
+ } else {
+ // Usage of DriverManager is not possible, as it does not
+ // respect the ContextClassLoader
+ driver = (Driver) driverFromCCL.newInstance();
+ if (!driver.acceptsURL(url)) {
+ throw new SQLException("No suitable driver", "08001");
+ }
+ }
+ } catch (Throwable t) {
+ String message = "Cannot create JDBC driver of class '" +
+ (driverClassName != null ? driverClassName : "") +
+ "' for connect URL '" + url + "'";
+ logWriter.println(message);
+ t.printStackTrace(logWriter);
+ throw new SQLNestedException(message, t);
+ }
+
+ // Can't test without a validationQuery
+ if (validationQuery == null) {
+ setTestOnBorrow(false);
+ setTestOnReturn(false);
+ setTestWhileIdle(false);
+ }
+
+ // Set up the driver connection factory we will use
+ String user = username;
+ if (user != null) {
+ connectionProperties.put("user", user);
+ } else {
+ log("DBCP DataSource configured without a 'username'");
+ }
+
+ String pwd = password;
+ if (pwd != null) {
+ connectionProperties.put("password", pwd);
+ } else {
+ log("DBCP DataSource configured without a 'password'");
+ }
+
+ ConnectionFactory driverConnectionFactory = new DriverConnectionFactory(driver, url, connectionProperties);
+ return driverConnectionFactory;
+ }
+}
http://git-wip-us.apache.org/repos/asf/oozie/blob/a54f7c20/core/src/main/java/org/apache/oozie/util/db/DatabaseRetryPredicate.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/oozie/util/db/DatabaseRetryPredicate.java b/core/src/main/java/org/apache/oozie/util/db/DatabaseRetryPredicate.java
new file mode 100644
index 0000000..c89aabe
--- /dev/null
+++ b/core/src/main/java/org/apache/oozie/util/db/DatabaseRetryPredicate.java
@@ -0,0 +1,51 @@
+/**
+ * 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.oozie.util.db;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import com.google.common.base.Predicate;
+
+public abstract class DatabaseRetryPredicate implements Predicate<Throwable> {
+
+ @Override
+ public abstract boolean apply(Throwable input);
+
+ /*
+ * Helper method for subclasses to retrieve all exceptions in a set. "All exceptions" means the exception
+ * hierarchy that can be walked by calling getCause() repeatedly.
+ *
+ * Subclasses either check if a particular exception was raised or not, or need the SQLException to extract the
+ * error code. In both cases the exception has to be found.
+ */
+ protected Set<Class<?>> getAllExceptions(final Throwable t) {
+ final Set<Class<?>> exceptions = new HashSet<>();
+
+ exceptions.add(t.getClass());
+
+ Throwable ex = t;
+ while (ex.getCause() != null) {
+ exceptions.add(ex.getCause().getClass());
+ ex = ex.getCause();
+ }
+
+ return exceptions;
+ }
+}
http://git-wip-us.apache.org/repos/asf/oozie/blob/a54f7c20/core/src/main/java/org/apache/oozie/util/db/FailingConnectionWrapper.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/oozie/util/db/FailingConnectionWrapper.java b/core/src/main/java/org/apache/oozie/util/db/FailingConnectionWrapper.java
new file mode 100644
index 0000000..0e31025
--- /dev/null
+++ b/core/src/main/java/org/apache/oozie/util/db/FailingConnectionWrapper.java
@@ -0,0 +1,371 @@
+/**
+ * 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.oozie.util.db;
+
+import com.google.common.base.Preconditions;
+import com.google.common.base.Predicate;
+import com.google.common.collect.Sets;
+import org.apache.directory.api.util.Strings;
+import org.apache.oozie.util.XLog;
+
+import javax.annotation.Nullable;
+import javax.persistence.PersistenceException;
+import java.sql.Array;
+import java.sql.Blob;
+import java.sql.CallableStatement;
+import java.sql.Clob;
+import java.sql.Connection;
+import java.sql.DatabaseMetaData;
+import java.sql.NClob;
+import java.sql.PreparedStatement;
+import java.sql.SQLClientInfoException;
+import java.sql.SQLException;
+import java.sql.SQLWarning;
+import java.sql.SQLXML;
+import java.sql.Savepoint;
+import java.sql.Statement;
+import java.sql.Struct;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.Executor;
+
+public class FailingConnectionWrapper implements Connection {
+ private static final XLog LOG = XLog.getLog(FailingConnectionWrapper.class);
+
+ private final Connection delegate;
+ private static final RuntimeExceptionInjector<PersistenceException> injector =
+ new RuntimeExceptionInjector<>(PersistenceException.class, 5);
+ private static final OozieDmlStatementPredicate oozieDmlStatementPredicate =
+ new OozieDmlStatementPredicate();
+
+ public FailingConnectionWrapper(final Connection delegate) throws SQLException {
+ this.delegate = delegate;
+ }
+
+ @Override
+ public Statement createStatement() throws SQLException {
+ return delegate.createStatement();
+ }
+
+ @Override
+ public PreparedStatement prepareStatement(final String sql) throws SQLException {
+ return delegate.prepareStatement(sql);
+ }
+
+ @Override
+ public CallableStatement prepareCall(final String sql) throws SQLException {
+ return delegate.prepareCall(sql);
+ }
+
+ @Override
+ public String nativeSQL(final String sql) throws SQLException {
+ return delegate.nativeSQL(sql);
+ }
+
+ @Override
+ public void setAutoCommit(final boolean autoCommit) throws SQLException {
+ delegate.setAutoCommit(autoCommit);
+ }
+
+ @Override
+ public boolean getAutoCommit() throws SQLException {
+ return delegate.getAutoCommit();
+ }
+
+ @Override
+ public void commit() throws SQLException {
+ delegate.commit();
+ }
+
+ @Override
+ public void rollback() throws SQLException {
+ delegate.rollback();
+ }
+
+ @Override
+ public void close() throws SQLException {
+ delegate.close();
+ }
+
+ @Override
+ public boolean isClosed() throws SQLException {
+ return delegate.isClosed();
+ }
+
+ @Override
+ public DatabaseMetaData getMetaData() throws SQLException {
+ return delegate.getMetaData();
+ }
+
+ @Override
+ public void setReadOnly(final boolean readOnly) throws SQLException {
+ delegate.setReadOnly(readOnly);
+ }
+
+ @Override
+ public boolean isReadOnly() throws SQLException {
+ return delegate.isReadOnly();
+ }
+
+ @Override
+ public void setCatalog(final String catalog) throws SQLException {
+ delegate.setCatalog(catalog);
+ }
+
+ @Override
+ public String getCatalog() throws SQLException {
+ return delegate.getCatalog();
+ }
+
+ @Override
+ public void setTransactionIsolation(final int level) throws SQLException {
+ delegate.setTransactionIsolation(level);
+ }
+
+ @Override
+ public int getTransactionIsolation() throws SQLException {
+ return delegate.getTransactionIsolation();
+ }
+
+ @Override
+ public SQLWarning getWarnings() throws SQLException {
+ return delegate.getWarnings();
+ }
+
+ @Override
+ public void clearWarnings() throws SQLException {
+ delegate.clearWarnings();
+ }
+
+ @Override
+ public Statement createStatement(final int resultSetType, final int resultSetConcurrency) throws SQLException {
+ return delegate.createStatement(resultSetType, resultSetConcurrency);
+ }
+
+ @Override
+ public PreparedStatement prepareStatement(final String sql, final int resultSetType, final int resultSetConcurrency)
+ throws SQLException {
+ if (oozieDmlStatementPredicate.apply(sql)) {
+ LOG.trace("Injecting random failure. It's a DML statement of an Oozie table, preparing this statement might fail.");
+ injector.inject(String.format("Deliberately failing to prepare statement. [sql=%s]", sql));
+ }
+
+ LOG.trace("Preparing statement. [sql={0}]", sql);
+ return delegate.prepareStatement(sql, resultSetType, resultSetConcurrency);
+ }
+
+ @Override
+ public CallableStatement prepareCall(final String sql, final int resultSetType, final int resultSetConcurrency)
+ throws SQLException {
+ return delegate.prepareCall(sql, resultSetType, resultSetConcurrency);
+ }
+
+ @Override
+ public Map<String, Class<?>> getTypeMap() throws SQLException {
+ return delegate.getTypeMap();
+ }
+
+ @Override
+ public void setTypeMap(final Map<String, Class<?>> map) throws SQLException {
+ delegate.setTypeMap(map);
+ }
+
+ @Override
+ public void setHoldability(final int holdability) throws SQLException {
+ delegate.setHoldability(holdability);
+ }
+
+ @Override
+ public int getHoldability() throws SQLException {
+ return delegate.getHoldability();
+ }
+
+ @Override
+ public Savepoint setSavepoint() throws SQLException {
+ return delegate.setSavepoint();
+ }
+
+ @Override
+ public Savepoint setSavepoint(final String name) throws SQLException {
+ return delegate.setSavepoint(name);
+ }
+
+ @Override
+ public void rollback(final Savepoint savepoint) throws SQLException {
+ delegate.rollback();
+ }
+
+ @Override
+ public void releaseSavepoint(final Savepoint savepoint) throws SQLException {
+ delegate.releaseSavepoint(savepoint);
+ }
+
+ @Override
+ public Statement createStatement(final int resultSetType, final int resultSetConcurrency, final int resultSetHoldability)
+ throws SQLException {
+ return delegate.createStatement(resultSetType, resultSetConcurrency, resultSetHoldability);
+ }
+
+ @Override
+ public PreparedStatement prepareStatement(final String sql, final int resultSetType, final int resultSetConcurrency,
+ final int resultSetHoldability) throws SQLException {
+ return delegate.prepareStatement(sql, resultSetType, resultSetConcurrency, resultSetHoldability);
+ }
+
+ @Override
+ public CallableStatement prepareCall(final String sql, final int resultSetType, final int resultSetConcurrency,
+ final int resultSetHoldability) throws SQLException {
+ return delegate.prepareCall(sql, resultSetType, resultSetConcurrency, resultSetHoldability);
+ }
+
+ @Override
+ public PreparedStatement prepareStatement(final String sql, final int autoGeneratedKeys) throws SQLException {
+ return delegate.prepareStatement(sql, autoGeneratedKeys);
+ }
+
+ @Override
+ public PreparedStatement prepareStatement(final String sql, final int[] columnIndexes) throws SQLException {
+ return delegate.prepareStatement(sql, columnIndexes);
+ }
+
+ @Override
+ public PreparedStatement prepareStatement(final String sql, final String[] columnNames) throws SQLException {
+ return delegate.prepareStatement(sql, columnNames);
+ }
+
+ @Override
+ public Clob createClob() throws SQLException {
+ return delegate.createClob();
+ }
+
+ @Override
+ public Blob createBlob() throws SQLException {
+ return delegate.createBlob();
+ }
+
+ @Override
+ public NClob createNClob() throws SQLException {
+ return delegate.createNClob();
+ }
+
+ @Override
+ public SQLXML createSQLXML() throws SQLException {
+ return delegate.createSQLXML();
+ }
+
+ @Override
+ public boolean isValid(final int timeout) throws SQLException {
+ return delegate.isValid(timeout);
+ }
+
+ @Override
+ public void setClientInfo(final String name, final String value) throws SQLClientInfoException {
+ delegate.setClientInfo(name, value);
+ }
+
+ @Override
+ public void setClientInfo(final Properties properties) throws SQLClientInfoException {
+ delegate.setClientInfo(properties);
+ }
+
+ @Override
+ public String getClientInfo(final String name) throws SQLException {
+ return delegate.getClientInfo(name);
+ }
+
+ @Override
+ public Properties getClientInfo() throws SQLException {
+ return delegate.getClientInfo();
+ }
+
+ @Override
+ public Array createArrayOf(final String typeName, final Object[] elements) throws SQLException {
+ return delegate.createArrayOf(typeName, elements);
+ }
+
+ @Override
+ public Struct createStruct(final String typeName, final Object[] attributes) throws SQLException {
+ return delegate.createStruct(typeName, attributes);
+ }
+
+ @Override
+ public void setSchema(final String schema) throws SQLException {
+ delegate.setSchema(schema);
+ }
+
+ @Override
+ public String getSchema() throws SQLException {
+ return delegate.getSchema();
+ }
+
+ @Override
+ public void abort(final Executor executor) throws SQLException {
+ delegate.abort(executor);
+ }
+
+ @Override
+ public void setNetworkTimeout(final Executor executor, final int milliseconds) throws SQLException {
+ delegate.setNetworkTimeout(executor, milliseconds);
+ }
+
+ @Override
+ public int getNetworkTimeout() throws SQLException {
+ return delegate.getNetworkTimeout();
+ }
+
+ @Override
+ public <T> T unwrap(final Class<T> iface) throws SQLException {
+ return delegate.unwrap(iface);
+ }
+
+ @Override
+ public boolean isWrapperFor(final Class<?> iface) throws SQLException {
+ return delegate.isWrapperFor(iface);
+ }
+
+ static class OozieDmlStatementPredicate implements Predicate<String> {
+ private static final Set<String> DML_PREFIXES = Sets.newHashSet(
+ "SELECT ", "INSERT INTO ", "UPDATE ", "DELETE FROM ");
+ private static final Set<String> OOZIE_TABLE_NAMES = Sets.newHashSet(
+ "BUNDLE_ACTIONS", "BUNDLE_JOBS", "COORD_ACTIONS", "COORD_JOBS", "SLA_REGISTRATION", "SLA_SUMMARY",
+ "WF_ACTIONS", "WF_JOBS");
+
+ @Override
+ public boolean apply(@Nullable String input) {
+ Preconditions.checkArgument(Strings.isNotEmpty(input));
+
+ boolean isDmlStatement = false;
+ for (final String dmlPrefix : DML_PREFIXES) {
+ if (input.toUpperCase().startsWith(dmlPrefix)) {
+ isDmlStatement = true;
+ }
+ }
+
+ boolean isOozieTable = false;
+ for (final String oozieTableName : OOZIE_TABLE_NAMES) {
+ if (input.toUpperCase().contains(oozieTableName)) {
+ isOozieTable = true;
+ }
+ }
+
+ return isDmlStatement && isOozieTable;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/oozie/blob/a54f7c20/core/src/main/java/org/apache/oozie/util/db/FailingHSQLDBDriverWrapper.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/oozie/util/db/FailingHSQLDBDriverWrapper.java b/core/src/main/java/org/apache/oozie/util/db/FailingHSQLDBDriverWrapper.java
new file mode 100644
index 0000000..fe9f08b
--- /dev/null
+++ b/core/src/main/java/org/apache/oozie/util/db/FailingHSQLDBDriverWrapper.java
@@ -0,0 +1,37 @@
+/**
+ * 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.oozie.util.db;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.util.Properties;
+
+public class FailingHSQLDBDriverWrapper extends org.hsqldb.jdbcDriver {
+
+ public static final String USE_FAILING_DRIVER = "oozie.sql.use.failing.driver";
+
+ public Connection connect(final String url,
+ final Properties info) throws SQLException {
+ if (Boolean.parseBoolean(System.getProperty(USE_FAILING_DRIVER, "false"))) {
+ return new FailingConnectionWrapper(super.connect(url, info));
+ }
+
+ return super.connect(url, info);
+ }
+}
http://git-wip-us.apache.org/repos/asf/oozie/blob/a54f7c20/core/src/main/java/org/apache/oozie/util/db/FailingMySQLDriverWrapper.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/oozie/util/db/FailingMySQLDriverWrapper.java b/core/src/main/java/org/apache/oozie/util/db/FailingMySQLDriverWrapper.java
new file mode 100644
index 0000000..f7af57c
--- /dev/null
+++ b/core/src/main/java/org/apache/oozie/util/db/FailingMySQLDriverWrapper.java
@@ -0,0 +1,51 @@
+/**
+ * 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.oozie.util.db;
+
+import com.mysql.jdbc.Driver;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.util.Properties;
+
+/**
+ * Used for runtime database error injection on MySQL.
+ * <p/>
+ * Necessary steps:
+ * <ul>
+ * <li>set {@code oozie.service.JPAService.connection.data.source} to
+ * {@code org.apache.oozie.util.db.BasicDataSourceWrapper} within {@code oozie-site.xml}</li>
+ * <li>set {@code oozie.service.JPAService.jdbc.driver} to {@code org.apache.oozie.util.db.FailingMySQLDriverWrapper}
+ * within {@code oozie-site.xml}</li>
+ * <li>restart Oozie server</li>
+ * <li>submit / start some workflows, coordinators etc.</li>
+ * <li>see any of those {@code JPAException} instances with following message prefix:
+ * {@code Deliberately failing to prepare statement.}</li>
+ * </ul>
+ */
+public class FailingMySQLDriverWrapper extends Driver {
+ public FailingMySQLDriverWrapper() throws SQLException {
+ super();
+ }
+
+ public Connection connect(final String url,
+ final Properties info) throws SQLException {
+ return new FailingConnectionWrapper(super.connect(url, info));
+ }
+}
http://git-wip-us.apache.org/repos/asf/oozie/blob/a54f7c20/core/src/main/java/org/apache/oozie/util/db/OperationRetryHandler.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/oozie/util/db/OperationRetryHandler.java b/core/src/main/java/org/apache/oozie/util/db/OperationRetryHandler.java
new file mode 100644
index 0000000..16a0a82
--- /dev/null
+++ b/core/src/main/java/org/apache/oozie/util/db/OperationRetryHandler.java
@@ -0,0 +1,130 @@
+/**
+ * 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.oozie.util.db;
+
+import java.util.concurrent.Callable;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.oozie.util.XLog;
+
+import com.google.common.base.Preconditions;
+import com.google.common.base.Predicate;
+
+public class OperationRetryHandler {
+ private static XLog LOG = XLog.getLog(OperationRetryHandler.class);
+ @VisibleForTesting
+ static final RetryAttemptState RETRY_ATTEMPT_STATE = new RetryAttemptState();
+
+ private final int maxRetryCount;
+ private final long initialWaitTime;
+ private final long maxWaitTime;
+ private final Predicate<Throwable> retryPredicate;
+ private final boolean shouldRetry;
+
+ public OperationRetryHandler(final int maxRetryCount, final long initialWaitTime, final long maxWaitTime,
+ final Predicate<Throwable> retryPredicate) {
+ Preconditions.checkArgument(maxRetryCount >= 0, "Retry count must not be less than zero");
+ Preconditions.checkArgument(initialWaitTime > 0, "Initial wait time must be greater than zero");
+ Preconditions.checkArgument(maxWaitTime >= 0, "Maximum wait time must not be less than zero");
+
+ this.maxRetryCount = maxRetryCount;
+ this.initialWaitTime = initialWaitTime;
+ this.maxWaitTime = maxWaitTime;
+ this.retryPredicate = Preconditions.checkNotNull(retryPredicate, "Retry predicate must not be null");
+ this.shouldRetry = !(maxRetryCount == 0 || maxWaitTime == 0);
+
+ LOG.trace("Retry handler parameters are set." +
+ "[maxRetryCount={0};initialWaitTime={1};maxWaitTime={2};retryPredicate.class={3};shouldRetry={4}]",
+ this.maxRetryCount, this.initialWaitTime, this.maxWaitTime, this.retryPredicate.getClass().getName(), shouldRetry);
+ }
+
+ public <V> V executeWithRetry(final Callable<V> operation) throws Exception {
+ int retries = 0;
+ long waitTime = initialWaitTime;
+ Exception lastException = null;
+
+ if (!shouldRetry) {
+ try {
+ LOG.trace("Configured not to retry, calling operation once.");
+
+ final V result = operation.call();
+
+ LOG.trace("Operation called once successfully.");
+
+ return result;
+ }
+ catch (final Exception e) {
+ LOG.error("An error occurred while calling the operation once. [e.message={0}]", e.getMessage());
+ throw e;
+ }
+ }
+
+ try {
+ RETRY_ATTEMPT_STATE.signalStart();
+
+ while (retries < maxRetryCount) {
+ try {
+ LOG.trace("Calling operation. [retries={0}]", retries);
+
+ retries++;
+ final V result = operation.call();
+
+ LOG.trace("Operation called successfully.");
+
+ return result;
+ } catch (final Exception e) {
+ LOG.warn("Database error", e);
+
+ // if retries have been done by an inner retry handler,
+ // then we won't make any effort to do it again
+ if (RETRY_ATTEMPT_STATE.isExhausted()) {
+ LOG.error("Retry attempts have been exhausted. [e.message={0}]", e.getMessage());
+ throw e;
+ }
+
+ if (retryPredicate.apply(e)) {
+ LOG.trace("Exception is not on blacklist, handling retry. [retries={0};e.class={1}]",
+ retries, e.getClass().getName());
+ waitTime = handleRetry(waitTime, retries);
+ lastException = e;
+ }
+ else {
+ LOG.warn("Exception is on blacklist, not handling retry. [retries={0};e.class={1}]",
+ retries, e.getClass().getName());
+ throw e;
+ }
+ }
+ }
+
+ LOG.error("Number of maximum retry attempts exhausted");
+ RETRY_ATTEMPT_STATE.signalExhausted(); // signal to possible outer retry handlers
+ throw lastException;
+ } finally {
+ RETRY_ATTEMPT_STATE.signalEnd();
+ }
+ }
+
+ private long handleRetry(long sleepBeforeRetryMs, final int retries) throws InterruptedException {
+ LOG.warn("Operation failed, sleeping {0} milliseconds before retry #{1}", sleepBeforeRetryMs, retries);
+ Thread.sleep(sleepBeforeRetryMs);
+ sleepBeforeRetryMs *= 2;
+
+ return sleepBeforeRetryMs > maxWaitTime ? maxWaitTime : sleepBeforeRetryMs;
+ }
+}
http://git-wip-us.apache.org/repos/asf/oozie/blob/a54f7c20/core/src/main/java/org/apache/oozie/util/db/PersistenceExceptionSubclassFilterRetryPredicate.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/oozie/util/db/PersistenceExceptionSubclassFilterRetryPredicate.java b/core/src/main/java/org/apache/oozie/util/db/PersistenceExceptionSubclassFilterRetryPredicate.java
new file mode 100644
index 0000000..b742ca7
--- /dev/null
+++ b/core/src/main/java/org/apache/oozie/util/db/PersistenceExceptionSubclassFilterRetryPredicate.java
@@ -0,0 +1,73 @@
+/**
+ * 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.oozie.util.db;
+
+import com.google.common.collect.Sets;
+import org.apache.oozie.util.XLog;
+
+import javax.persistence.EntityExistsException;
+import javax.persistence.EntityNotFoundException;
+import javax.persistence.LockTimeoutException;
+import javax.persistence.NoResultException;
+import javax.persistence.NonUniqueResultException;
+import javax.persistence.OptimisticLockException;
+import javax.persistence.PersistenceException;
+import javax.persistence.PessimisticLockException;
+import javax.persistence.QueryTimeoutException;
+import javax.persistence.TransactionRequiredException;
+import java.util.Set;
+
+/**
+ * A {@link DatabaseRetryPredicate} which applies when a given {@link Exception} (or its causes) are NOT blacklisted.
+ * <p/>
+ * Blacklisted exceptions in this class do not indicate a network failure, therefore no retry should take place.
+ */
+public class PersistenceExceptionSubclassFilterRetryPredicate extends DatabaseRetryPredicate {
+ private static final XLog LOG = XLog.getLog(PersistenceExceptionSubclassFilterRetryPredicate.class);
+ private static final Set<Class<? extends PersistenceException>> BLACKLIST = Sets.newHashSet(
+ EntityExistsException.class,
+ EntityNotFoundException.class,
+ LockTimeoutException.class,
+ NoResultException.class,
+ NonUniqueResultException.class,
+ OptimisticLockException.class,
+ PessimisticLockException.class,
+ QueryTimeoutException.class,
+ TransactionRequiredException.class
+ );
+
+ @Override
+ public boolean apply(final Throwable throwable) {
+ LOG.trace("Retry predicate investigation started. [throwable.class={0}]", throwable.getClass().getName());
+
+ boolean applies = true;
+
+ for (final Class<?> classDownTheStackTrace : getAllExceptions(throwable)) {
+ for (final Class<? extends PersistenceException> blacklistElement : BLACKLIST) {
+ if (blacklistElement.isAssignableFrom(classDownTheStackTrace)) {
+ applies = false;
+ }
+ }
+ }
+
+ LOG.trace("Retry predicate investigation finished. [applies={0}]", applies);
+
+ return applies;
+ }
+}
http://git-wip-us.apache.org/repos/asf/oozie/blob/a54f7c20/core/src/main/java/org/apache/oozie/util/db/RetryAttemptState.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/oozie/util/db/RetryAttemptState.java b/core/src/main/java/org/apache/oozie/util/db/RetryAttemptState.java
new file mode 100644
index 0000000..0413ee0
--- /dev/null
+++ b/core/src/main/java/org/apache/oozie/util/db/RetryAttemptState.java
@@ -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.oozie.util.db;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+
+/**
+ * This class tracks nested {@link OperationRetryHandler} calls. Some {@code JPAExecutor} implementations call other
+ * {@code JPAExecutor}s. This results in two (or possibly more) {@link OperationRetryHandler#executeWithRetry(Callable)} calls.
+ * <p/>
+ * If the innermost retry handler has exhausted all attempts and rethrows the {@link Exception}, the outer handler catches that
+ * and would restart the JPA operation again.
+ * <p/>
+ * In order to avoid this, retry handlers must communicate with each other on the same thread by incrementing / decrementing the
+ * nesting level and signalling whether the maximum number of attempts have been reached.
+ * <p/>
+ * We use {@link ThreadLocal}s because retry handlers might be called from different threads in parallel. If the nesting level is
+ * zero, it's important to reset the {@link #exhausted} back to {@code false} since this variable is reused in the
+ * thread pool.
+ */
+final class RetryAttemptState {
+ private final ThreadLocal<Boolean> exhausted = new ThreadLocal<Boolean>() {
+ @Override
+ protected Boolean initialValue() {
+ return Boolean.FALSE;
+ }
+ };
+
+ private final ThreadLocal<Integer> inProgressCount = new ThreadLocal<Integer>() {
+ @Override
+ protected Integer initialValue() {
+ return 0;
+ }
+ };
+
+ void signalStart() {
+ Preconditions.checkState(!isExhausted(), "retry attempts exhausted");
+
+ inProgressCount.set(inProgressCount.get() + 1);
+ }
+
+ void signalEnd() {
+ int currentLevel = inProgressCount.get() - 1;
+ inProgressCount.set(currentLevel);
+ if (currentLevel == 0) {
+ // state must be reset
+ exhausted.set(false);
+ }
+ }
+
+ void signalExhausted() {
+ exhausted.set(true);
+ }
+
+ boolean isExhausted() {
+ return exhausted.get();
+ }
+
+ @VisibleForTesting
+ int getInProgressCount() {
+ return inProgressCount.get();
+ }
+}
http://git-wip-us.apache.org/repos/asf/oozie/blob/a54f7c20/core/src/main/java/org/apache/oozie/util/db/RuntimeExceptionInjector.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/oozie/util/db/RuntimeExceptionInjector.java b/core/src/main/java/org/apache/oozie/util/db/RuntimeExceptionInjector.java
new file mode 100644
index 0000000..2b55e85
--- /dev/null
+++ b/core/src/main/java/org/apache/oozie/util/db/RuntimeExceptionInjector.java
@@ -0,0 +1,77 @@
+/**
+ * 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.oozie.util.db;
+
+import com.google.common.base.Preconditions;
+import org.apache.oozie.util.XLog;
+
+import java.lang.reflect.InvocationTargetException;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.atomic.AtomicLong;
+
+public class RuntimeExceptionInjector<E extends RuntimeException> {
+ private static final XLog LOG = XLog.getLog(RuntimeExceptionInjector.class);
+ private static final AtomicLong failureCounter = new AtomicLong(0);
+
+ private final Class<E> runtimeExceptionClass;
+ private final int failurePercent;
+
+ public RuntimeExceptionInjector(final Class<E> runtimeExceptionClass, final int failurePercent) {
+ Preconditions.checkArgument(failurePercent <= 100 && failurePercent >= 0,
+ "illegal value for failure %: " + failurePercent);
+
+ this.runtimeExceptionClass = runtimeExceptionClass;
+ this.failurePercent = failurePercent;
+ }
+
+ public void inject(final String errorMessage) {
+ LOG.trace("Trying to inject random failure. [errorMessage={0}]", errorMessage);
+
+ final ThreadLocalRandom random = ThreadLocalRandom.current();
+ final int randomVal = random.nextInt(0, 100); // range: [0..99]
+
+ if (randomVal < failurePercent) {
+ final long count = failureCounter.incrementAndGet();
+ LOG.warn("Injecting random failure. [runtimeExceptionClass.name={0};count={1};errorMessage={2}]",
+ runtimeExceptionClass.getName(), count, errorMessage);
+ E injected;
+
+ try {
+ injected = runtimeExceptionClass.getConstructor(String.class).newInstance(
+ "injected random failure #" + count + " ." + errorMessage);
+ } catch (final InstantiationException | IllegalAccessException | InvocationTargetException
+ | NoSuchMethodException outer) {
+ try {
+ LOG.warn("Instantiating without error message. [runtimeExceptionClass.name={0};outer.message={1}]",
+ runtimeExceptionClass.getName(), outer.getMessage());
+ injected = runtimeExceptionClass.newInstance();
+ } catch (final InstantiationException | IllegalAccessException inner) {
+ LOG.error("Could not instantiate. [runtimeExceptionClass.name={0};inner.message={1}]",
+ runtimeExceptionClass.getName(), inner.getMessage());
+ throw new RuntimeException(inner);
+ }
+
+ }
+
+ throw injected;
+ }
+
+ LOG.trace("Did not inject random failure.");
+ }
+}
http://git-wip-us.apache.org/repos/asf/oozie/blob/a54f7c20/core/src/main/resources/META-INF/persistence.xml
----------------------------------------------------------------------
diff --git a/core/src/main/resources/META-INF/persistence.xml b/core/src/main/resources/META-INF/persistence.xml
index bad9278..a74078a 100644
--- a/core/src/main/resources/META-INF/persistence.xml
+++ b/core/src/main/resources/META-INF/persistence.xml
@@ -43,7 +43,7 @@
<class>org.apache.oozie.util.db.ValidateConnectionBean</class>
<properties>
- <property name="openjpa.ConnectionDriverName" value="org.apache.commons.dbcp.BasicDataSource"/>
+ <property name="openjpa.ConnectionDriverName" value="org.apache.oozie.util.db.BasicDataSourceWrapper"/>
<property name="openjpa.ConnectionProperties" value="**INVALID**"/> <!--Set by StoreService at init time -->
@@ -102,7 +102,7 @@
<class>org.apache.oozie.util.db.ValidateConnectionBean</class>
<properties>
- <property name="openjpa.ConnectionDriverName" value="org.apache.commons.dbcp.BasicDataSource"/>
+ <property name="openjpa.ConnectionDriverName" value="org.apache.oozie.util.db.BasicDataSourceWrapper"/>
<property name="openjpa.ConnectionProperties" value="**INVALID**"/> <!--Set by StoreService at init time -->
@@ -163,7 +163,7 @@
<class>org.apache.oozie.util.db.ValidateConnectionBean</class>
<properties>
- <property name="openjpa.ConnectionDriverName" value="org.apache.commons.dbcp.BasicDataSource"/>
+ <property name="openjpa.ConnectionDriverName" value="org.apache.oozie.util.db.BasicDataSourceWrapper"/>
<property name="openjpa.ConnectionProperties" value="**INVALID**"/> <!--Set by StoreService at init time -->
@@ -222,7 +222,7 @@
<class>org.apache.oozie.util.db.ValidateConnectionBean</class>
<properties>
- <property name="openjpa.ConnectionDriverName" value="org.apache.commons.dbcp.BasicDataSource"/>
+ <property name="openjpa.ConnectionDriverName" value="org.apache.oozie.util.db.BasicDataSourceWrapper"/>
<property name="openjpa.ConnectionProperties" value="**INVALID**"/> <!--Set by StoreService at init time -->
@@ -282,7 +282,7 @@
<class>org.apache.oozie.util.db.ValidateConnectionBean</class>
<properties>
- <property name="openjpa.ConnectionDriverName" value="org.apache.commons.dbcp.BasicDataSource"/>
+ <property name="openjpa.ConnectionDriverName" value="org.apache.oozie.util.db.BasicDataSourceWrapper"/>
<property name="openjpa.ConnectionProperties" value="**INVALID**"/> <!--Set by StoreService at init time -->
@@ -342,7 +342,7 @@
<class>org.apache.oozie.util.db.ValidateConnectionBean</class>
<properties>
- <property name="openjpa.ConnectionDriverName" value="org.apache.commons.dbcp.BasicDataSource"/>
+ <property name="openjpa.ConnectionDriverName" value="org.apache.oozie.util.db.BasicDataSourceWrapper"/>
<property name="openjpa.ConnectionProperties" value="**INVALID**"/> <!--Set by StoreService at init time -->