diff --git a/integration-tests/transaction/src/test/java/org/apache/batchee/its/transaction/TxErrorTest.java b/integration-tests/transaction/src/test/java/org/apache/batchee/its/transaction/TxErrorTest.java
index acdbd51..564c7a7 100644
--- a/integration-tests/transaction/src/test/java/org/apache/batchee/its/transaction/TxErrorTest.java
+++ b/integration-tests/transaction/src/test/java/org/apache/batchee/its/transaction/TxErrorTest.java
@@ -19,6 +19,10 @@
 import javax.batch.operations.JobOperator;
 import javax.batch.runtime.BatchRuntime;
 import javax.batch.runtime.BatchStatus;
+import javax.batch.runtime.Metric;
+import javax.batch.runtime.StepExecution;
+
+import java.util.List;
 
 import org.apache.batchee.util.Batches;
 import org.testng.Assert;
@@ -30,8 +34,27 @@
     @Test
     public void testRolledBackDuringWork() {
         final JobOperator jobOperator = BatchRuntime.getJobOperator();
-        BatchStatus batchStatus = Batches.waitFor(jobOperator, jobOperator.start("txtest1", null));
-        Assert.assertEquals(batchStatus, BatchStatus.COMPLETED);
-        Assert.assertEquals(TxErrorWriter1.written.intValue(), 5);
+        long executionId = jobOperator.start("txtest1", null);
+        BatchStatus batchStatus = Batches.waitFor(jobOperator, executionId);
+        Assert.assertEquals(batchStatus, BatchStatus.FAILED);
+        Assert.assertEquals(TxErrorWriter1.written.intValue(), 3);
+
+        List<StepExecution> stepExecutions = jobOperator.getStepExecutions(executionId);
+        Assert.assertEquals(stepExecutions.size(), 1);
+        StepExecution stepExecution = stepExecutions.get(0);
+        Metric[] metrics = stepExecution.getMetrics();
+        assertMetric(Metric.MetricType.READ_COUNT, 2, metrics);
+        assertMetric(Metric.MetricType.WRITE_COUNT, 2, metrics);
+        assertMetric(Metric.MetricType.ROLLBACK_COUNT, 1, metrics);
+    }
+
+    private void assertMetric(Metric.MetricType metricType, long expected, Metric[] metrics) {
+        for (Metric metric : metrics) {
+            if (metricType.equals(metric.getType())) {
+                Assert.assertEquals(metric.getValue(), expected);
+                return;
+            }
+        }
+        Assert.fail("MetricType " + metricType + " not in collected metrics");
     }
 }
diff --git a/integration-tests/transaction/src/test/resources/META-INF/batch-jobs/txtest1.xml b/integration-tests/transaction/src/test/resources/META-INF/batch-jobs/txtest1.xml
index 4b7a67c..ba6809f 100644
--- a/integration-tests/transaction/src/test/resources/META-INF/batch-jobs/txtest1.xml
+++ b/integration-tests/transaction/src/test/resources/META-INF/batch-jobs/txtest1.xml
@@ -18,9 +18,6 @@
         <chunk item-count="1">
             <reader ref="org.apache.batchee.its.transaction.TxErrorReader"></reader>
             <writer ref="org.apache.batchee.its.transaction.TxErrorWriter1"></writer>
-            <skippable-exception-classes>
-                <include class="java.lang.Exception"/> <!-- all exceptions... -->
-            </skippable-exception-classes>
         </chunk>
     </step>
 </job>
diff --git a/jbatch/src/main/java/org/apache/batchee/container/impl/controller/BaseStepController.java b/jbatch/src/main/java/org/apache/batchee/container/impl/controller/BaseStepController.java
index d07d858..63381aa 100755
--- a/jbatch/src/main/java/org/apache/batchee/container/impl/controller/BaseStepController.java
+++ b/jbatch/src/main/java/org/apache/batchee/container/impl/controller/BaseStepController.java
@@ -359,6 +359,11 @@
     }
 
     protected void persistUserData() {
+        PersistentDataWrapper userData = resolveUserData();
+        storeUserData(userData);
+    }
+
+    protected PersistentDataWrapper resolveUserData() {
         final ByteArrayOutputStream persistentBAOS = new ByteArrayOutputStream();
         final ObjectOutputStream persistentDataOOS;
 
@@ -370,8 +375,16 @@
             throw new BatchContainerServiceException("Cannot persist the persistent user data for the step.", e);
         }
 
-        stepStatus.setPersistentUserData(new PersistentDataWrapper(persistentBAOS.toByteArray()));
-        statusManagerService.updateStepStatus(stepStatus.getStepExecutionId(), stepStatus);
+        return new PersistentDataWrapper(persistentBAOS.toByteArray());
+    }
+
+    protected void storeUserData(PersistentDataWrapper userData) {
+        try {
+            stepStatus.setPersistentUserData(userData);
+            statusManagerService.updateStepStatus(stepStatus.getStepExecutionId(), stepStatus);
+        } catch (final Exception e) {
+            throw new BatchContainerServiceException("Cannot persist the persistent user data for the step.", e);
+        }
     }
 
     protected void persistExitStatusAndEndTimestamp() {
diff --git a/jbatch/src/main/java/org/apache/batchee/container/impl/controller/chunk/CheckpointManager.java b/jbatch/src/main/java/org/apache/batchee/container/impl/controller/chunk/CheckpointManager.java
index 7e07450..584d737 100755
--- a/jbatch/src/main/java/org/apache/batchee/container/impl/controller/chunk/CheckpointManager.java
+++ b/jbatch/src/main/java/org/apache/batchee/container/impl/controller/chunk/CheckpointManager.java
@@ -16,6 +16,9 @@
 */
 package org.apache.batchee.container.impl.controller.chunk;
 
+import java.util.HashMap;
+import java.util.Map;
+
 import org.apache.batchee.container.exception.BatchContainerRuntimeException;
 import org.apache.batchee.container.exception.BatchContainerServiceException;
 import org.apache.batchee.spi.DataRepresentationService;
@@ -75,28 +78,45 @@
         }
     }
 
-    public void checkpoint() {
+    /**
+     * Takes the current checkpoint data from the ItemReader and ItemWriter
+     * and store them in the database
+     */
+    public Map<CheckpointDataKey, CheckpointData> prepareCheckpoints() {
         final CheckpointDataKey readerChkptDK;
         final CheckpointDataKey writerChkptDK;
+        Map<CheckpointDataKey, CheckpointData> checkpoints = new HashMap<CheckpointDataKey, CheckpointData>(2);
         try {
             byte[] checkpointBytes = dataRepresentationService.toInternalRepresentation(readerProxy.checkpointInfo());
             CheckpointData readerChkptData = new CheckpointData(jobInstanceID, stepId, CheckpointType.READER);
             readerChkptData.setRestartToken(checkpointBytes);
             readerChkptDK = new CheckpointDataKey(jobInstanceID, stepId, CheckpointType.READER);
 
-            persistenceManagerService.setCheckpointData(readerChkptDK, readerChkptData);
+            checkpoints.put(readerChkptDK, readerChkptData);
 
             checkpointBytes = dataRepresentationService.toInternalRepresentation(writerProxy.checkpointInfo());
             CheckpointData writerChkptData = new CheckpointData(jobInstanceID, stepId, CheckpointType.WRITER);
             writerChkptData.setRestartToken(checkpointBytes);
             writerChkptDK = new CheckpointDataKey(jobInstanceID, stepId, CheckpointType.WRITER);
 
-            persistenceManagerService.setCheckpointData(writerChkptDK, writerChkptData);
-
+            checkpoints.put(writerChkptDK, writerChkptData);
         } catch (final Exception ex) {
             // is this what I should be throwing here?
             throw new BatchContainerServiceException("Cannot persist the checkpoint data for [" + stepId + "]", ex);
         }
+
+        return checkpoints;
+    }
+
+    public void storeCheckPoints(Map<CheckpointDataKey, CheckpointData> checkpoints) {
+        try {
+            for (Map.Entry<CheckpointDataKey, CheckpointData> checkpointEntry : checkpoints.entrySet()) {
+                persistenceManagerService.setCheckpointData(checkpointEntry.getKey(), checkpointEntry.getValue());
+            }
+        } catch (final Exception ex) {
+            throw new BatchContainerServiceException("Cannot persist the checkpoint data for [" + stepId + "]", ex);
+        }
+
     }
 
     public int checkpointTimeout() {
diff --git a/jbatch/src/main/java/org/apache/batchee/container/impl/controller/chunk/ChunkStepController.java b/jbatch/src/main/java/org/apache/batchee/container/impl/controller/chunk/ChunkStepController.java
index ba256fa..6d72ccd 100755
--- a/jbatch/src/main/java/org/apache/batchee/container/impl/controller/chunk/ChunkStepController.java
+++ b/jbatch/src/main/java/org/apache/batchee/container/impl/controller/chunk/ChunkStepController.java
@@ -48,10 +48,14 @@
 import javax.batch.api.chunk.listener.SkipProcessListener;
 import javax.batch.api.chunk.listener.SkipReadListener;
 import javax.batch.api.chunk.listener.SkipWriteListener;
+import javax.batch.operations.BatchRuntimeException;
 import javax.batch.runtime.BatchStatus;
+import javax.transaction.Status;
+
 import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import java.util.Properties;
 import java.util.concurrent.BlockingQueue;
 import java.util.logging.Level;
@@ -585,11 +589,22 @@
                     chunkProxy.afterChunk();
                 }
 
-                checkpointManager.checkpoint();
-
-                this.persistUserData();
-
-                transactionManager.commit();
+                Map<CheckpointDataKey, CheckpointData> checkpoints = checkpointManager.prepareCheckpoints();
+                PersistentDataWrapper userData = resolveUserData();
+                try {
+                    transactionManager.commit();
+                    storeUserData(userData);
+                    checkpointManager.storeCheckPoints(checkpoints);
+                } catch (Exception e) {
+                    // only set the Exception if we didn't blow up before anyway
+                    if (this.stepContext.getException() != null) {
+                        this.stepContext.setException(e);
+                    }
+                    if (e instanceof BatchRuntimeException) {
+                        throw e;
+                    }
+                    throw new BatchContainerServiceException("Cannot commit the transaction for the step.", e);
+                }
 
                 checkpointManager.endCheckpoint();
 
@@ -659,14 +674,15 @@
      */
     private void rollback(final Throwable t) {
         try {
-            // ignore, we blow up anyway
-            transactionManager.setRollbackOnly();
             try {
                 doClose();
             } catch (Exception e) {
                 // ignore, we blow up anyway
             }
 
+            // ignore, we blow up anyway
+            transactionManager.setRollbackOnly();
+
             if (t instanceof Exception) {
                 Exception e = (Exception) t;
                 for (ChunkListener chunkProxy : chunkListeners) {
@@ -681,7 +697,10 @@
             // ever come up in the spec, but seems marginally more useful.
             stepContext.getMetric(MetricImpl.MetricType.ROLLBACK_COUNT).incValue();
         } finally {
-            transactionManager.rollback();
+            int txStatus = transactionManager.getStatus();
+            if (txStatus == Status.STATUS_ACTIVE || txStatus == Status.STATUS_MARKED_ROLLBACK) {
+                transactionManager.rollback();
+            }
             throw new BatchContainerRuntimeException("Failure in Read-Process-Write Loop", t);
         }
     }