3770 fix batch2 failure loop (#3779)

* 3770 fixing error handling

* some updates to error handling

* changelog

* blah

* bumping version

* fix test failures

* code review points

* review points

* add retry test

* assert counts in the test

* test fixes

Co-authored-by: leif stawnyczy <leifstawnyczy@leifs-MacBook-Pro.local>
Co-authored-by: Ken Stevens <ken@smilecdr.com>

hapi-deployable-pom/pom.xml

@@ -4,7 +4,7 @@
     <parent>
 		 <groupId>ca.uhn.hapi.fhir</groupId>
 		 <artifactId>hapi-fhir</artifactId>
-		 <version>6.1.0-PRE13-SNAPSHOT</version>
+		 <version>6.1.0-PRE14-SNAPSHOT</version>
 		 <relativePath>../pom.xml</relativePath>
 	 </parent>
 

hapi-fhir-android/pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
 

hapi-fhir-base/pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
 

hapi-fhir-batch/pom.xml

@@ -6,7 +6,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
 

hapi-fhir-bom/pom.xml

@@ -3,14 +3,14 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>ca.uhn.hapi.fhir</groupId>
     <artifactId>hapi-fhir-bom</artifactId>
-    <version>6.1.0-PRE13-SNAPSHOT</version>
+    <version>6.1.0-PRE14-SNAPSHOT</version>
     <packaging>pom</packaging>
 	 <name>HAPI FHIR BOM</name>
 
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
     

hapi-fhir-checkstyle/pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-fhir</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

hapi-fhir-cli/hapi-fhir-cli-api/pom.xml

@@ -4,7 +4,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
 

hapi-fhir-cli/hapi-fhir-cli-app/pom.xml

@@ -6,7 +6,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-fhir-cli</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

hapi-fhir-cli/hapi-fhir-cli-jpaserver/pom.xml

@@ -6,7 +6,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../../hapi-deployable-pom</relativePath>
 	</parent>
 

hapi-fhir-cli/pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-fhir</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

hapi-fhir-client-okhttp/pom.xml

@@ -4,7 +4,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
 

hapi-fhir-client/pom.xml

@@ -4,7 +4,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
 

hapi-fhir-converter/pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
 

hapi-fhir-dist/pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-fhir</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

hapi-fhir-docs/pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
 

hapi-fhir-docs/src/main/resources/ca/uhn/hapi/fhir/changelog/6_1_0/3770-added-error-handling-for-batch2.yaml

@@ -0,0 +1,6 @@
+---
+type: fix
+issue: 3770
+title: "Batch2 will have a standard error handling that will fail the job if it fails
+a chunk processing for more than 3 times.
+Further, added better validation to reindex job to disallow bad urls."

hapi-fhir-jacoco/pom.xml

@@ -11,7 +11,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
 

hapi-fhir-jaxrsserver-base/pom.xml

@@ -4,7 +4,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
 

hapi-fhir-jpa/pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
 	<modelVersion>4.0.0</modelVersion>

hapi-fhir-jpaserver-base/pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
 

hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/batch2/JpaJobPersistenceImpl.java

@@ -24,6 +24,7 @@ import ca.uhn.fhir.batch2.api.IJobPersistence;
 import ca.uhn.fhir.batch2.api.JobOperationResultJson;
 import ca.uhn.fhir.batch2.coordinator.BatchWorkChunk;
 import ca.uhn.fhir.batch2.model.JobInstance;
+import ca.uhn.fhir.batch2.model.MarkWorkChunkAsErrorRequest;
 import ca.uhn.fhir.batch2.model.StatusEnum;
 import ca.uhn.fhir.batch2.model.WorkChunk;
 import ca.uhn.fhir.jpa.dao.data.IBatch2JobInstanceRepository;
@@ -198,6 +199,14 @@ public class JpaJobPersistenceImpl implements IJobPersistence {
 		myWorkChunkRepository.updateChunkStatusAndIncrementErrorCountForEndError(theChunkId, new Date(), theErrorMessage, StatusEnum.ERRORED);
 	}
 
+	@Override
+	public Optional<WorkChunk> markWorkChunkAsErroredAndIncrementErrorCount(MarkWorkChunkAsErrorRequest theParameters) {
+		markWorkChunkAsErroredAndIncrementErrorCount(theParameters.getChunkId(), theParameters.getErrorMsg());
+		Optional<Batch2WorkChunkEntity> op = myWorkChunkRepository.findById(theParameters.getChunkId());
+
+		return op.map(c -> toChunk(c, theParameters.isIncludeData()));
+	}
+
 	@Override
 	public void markWorkChunkAsFailed(String theChunkId, String theErrorMessage) {
 		myWorkChunkRepository.updateChunkStatusAndIncrementErrorCountForEndError(theChunkId, new Date(), theErrorMessage, StatusEnum.FAILED);

hapi-fhir-jpaserver-base/src/test/java/ca/uhn/fhir/jpa/batch2/JpaJobPersistenceImplTest.java

@@ -27,6 +27,7 @@ class JpaJobPersistenceImplTest {
 	@InjectMocks
 	JpaJobPersistenceImpl mySvc;
 
+
 	@Test
 	void cancelSuccess() {
 		// setup

hapi-fhir-jpaserver-cql/pom.xml

@@ -7,7 +7,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
 

hapi-fhir-jpaserver-mdm/pom.xml

@@ -6,7 +6,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
 

hapi-fhir-jpaserver-model/pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
 

hapi-fhir-jpaserver-searchparam/pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
 

hapi-fhir-jpaserver-subscription/pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
 

hapi-fhir-jpaserver-test-utilities/pom.xml

@@ -6,7 +6,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
 

hapi-fhir-jpaserver-test-utilities/src/main/java/ca/uhn/fhir/jpa/test/Batch2JobHelper.java

@@ -28,6 +28,9 @@ import org.awaitility.core.ConditionTimeoutException;
 import org.hamcrest.Matchers;
 import org.springframework.beans.factory.annotation.Autowired;
 
+import java.time.Duration;
+import java.util.concurrent.TimeUnit;
+
 import java.util.Collection;
 import java.util.List;
 
@@ -71,6 +74,17 @@ public class Batch2JobHelper {
 		}, equalTo(StatusEnum.CANCELLED));
 	}
 
+	public JobInstance awaitJobHitsStatusInTime(String theId, int theSeconds, StatusEnum... theStatuses) {
+		await().atMost(theSeconds, TimeUnit.SECONDS)
+			.pollDelay(Duration.ofSeconds(10))
+			.until(() -> {
+				myJobMaintenanceService.runMaintenancePass();
+				return myJobCoordinator.getInstance(theId).getStatus();
+			}, Matchers.in(theStatuses));
+
+		return myJobCoordinator.getInstance(theId);
+	}
+
 	public void awaitJobInProgress(String theId) {
 		await().until(() -> {
 			myJobMaintenanceService.runMaintenancePass();

hapi-fhir-jpaserver-test-utilities/src/test/java/ca/uhn/fhir/jpa/batch2/Batch2CoordinatorIT.java

@@ -7,6 +7,7 @@ import ca.uhn.fhir.batch2.api.IJobDataSink;
 import ca.uhn.fhir.batch2.api.IJobMaintenanceService;
 import ca.uhn.fhir.batch2.api.IJobPersistence;
 import ca.uhn.fhir.batch2.api.IJobStepWorker;
+import ca.uhn.fhir.batch2.api.ILastJobStepWorker;
 import ca.uhn.fhir.batch2.api.IReductionStepWorker;
 import ca.uhn.fhir.batch2.api.JobExecutionFailedException;
 import ca.uhn.fhir.batch2.api.RunOutcome;
@@ -17,6 +18,11 @@ import ca.uhn.fhir.batch2.model.ChunkOutcome;
 import ca.uhn.fhir.batch2.model.JobDefinition;
 import ca.uhn.fhir.batch2.model.JobInstance;
 import ca.uhn.fhir.batch2.model.JobInstanceStartRequest;
+import ca.uhn.fhir.batch2.model.JobWorkNotificationJsonMessage;
+import ca.uhn.fhir.batch2.model.StatusEnum;
+import ca.uhn.fhir.jpa.subscription.channel.api.ChannelConsumerSettings;
+import ca.uhn.fhir.jpa.subscription.channel.api.IChannelFactory;
+import ca.uhn.fhir.jpa.subscription.channel.impl.LinkedBlockingChannel;
 import ca.uhn.fhir.jpa.test.BaseJpaR4Test;
 import ca.uhn.fhir.jpa.test.Batch2JobHelper;
 import ca.uhn.fhir.model.api.IModelJson;
@@ -24,11 +30,16 @@ import ca.uhn.fhir.util.JsonUtil;
 import ca.uhn.test.concurrency.LatchTimedOutError;
 import ca.uhn.test.concurrency.PointcutLatch;
 import com.fasterxml.jackson.annotation.JsonProperty;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.support.ExecutorChannelInterceptor;
 
 import javax.annotation.Nonnull;
 import java.util.ArrayList;
@@ -39,6 +50,8 @@ import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import static ca.uhn.fhir.batch2.config.BaseBatch2Config.CHANNEL_NAME;
+import static ca.uhn.fhir.batch2.coordinator.StepExecutionSvc.MAX_CHUNK_ERROR_COUNT;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -58,6 +71,8 @@ public class Batch2CoordinatorIT extends BaseJpaR4Test {
 	IJobMaintenanceService myJobMaintenanceService;
 	@Autowired
 	Batch2JobHelper myBatch2JobHelper;
+	@Autowired
+	private IChannelFactory myChannelFactory;
 
 	@Autowired
 	IJobPersistence myJobPersistence;
@@ -65,6 +80,7 @@ public class Batch2CoordinatorIT extends BaseJpaR4Test {
 	private final PointcutLatch myFirstStepLatch = new PointcutLatch("First Step");
 	private final PointcutLatch myLastStepLatch = new PointcutLatch("Last Step");
 	private IJobCompletionHandler<TestJobParameters> myCompletionHandler;
+	private LinkedBlockingChannel myWorkChannel;
 
 	private static RunOutcome callLatch(PointcutLatch theLatch, StepExecutionDetails<?, ?> theStep) {
 		theLatch.call(theStep);
@@ -74,6 +90,12 @@ public class Batch2CoordinatorIT extends BaseJpaR4Test {
 	@BeforeEach
 	public void before() {
 		myCompletionHandler = details -> {};
+		myWorkChannel = (LinkedBlockingChannel) myChannelFactory.getOrCreateReceiver(CHANNEL_NAME, JobWorkNotificationJsonMessage.class, new ChannelConsumerSettings());
+	}
+
+	@AfterEach
+	public void after() {
+		myWorkChannel.clearInterceptorsForUnitTest();
 	}
 
 	@Test
@@ -123,7 +145,6 @@ public class Batch2CoordinatorIT extends BaseJpaR4Test {
 		myLastStepLatch.awaitExpected();
 	}
 
-
 	@Test
 	public void testFastTrack_Maintenance_do_not_both_call_CompletionHandler() throws InterruptedException {
 		IJobStepWorker<TestJobParameters, VoidModel, FirstStepOutput> firstStep = (step, sink) -> {
@@ -189,7 +210,6 @@ public class Batch2CoordinatorIT extends BaseJpaR4Test {
 		myLastStepLatch.awaitExpected();
 	}
 
-
 	@Test
 	public void testJobDefinitionWithReductionStepIT() throws InterruptedException {
 		// setup
@@ -371,6 +391,69 @@ public class Batch2CoordinatorIT extends BaseJpaR4Test {
 		myBatch2JobHelper.awaitJobCancelled(instanceId);
 	}
 
+	@Test
+	public void testStepRunFailure_continuouslyThrows_marksJobFailed() {
+		AtomicInteger interceptorCounter = new AtomicInteger();
+		myWorkChannel.addInterceptor(new ExecutorChannelInterceptor() {
+			@Override
+			public void afterMessageHandled(Message<?> message, MessageChannel channel, MessageHandler handler, Exception ex) {
+				if (ex != null) {
+					interceptorCounter.incrementAndGet();
+					ourLog.info("Work Channel Exception thrown: {}.  Resending message", ex.getMessage());
+					channel.send(message);
+				}
+			}
+		});
+
+		// setup
+		AtomicInteger counter = new AtomicInteger();
+		// step 1
+		IJobStepWorker<TestJobParameters, VoidModel, FirstStepOutput> first = (step, sink) -> {
+			counter.getAndIncrement();
+			throw new RuntimeException("Exception");
+		};
+		// final step
+		ILastJobStepWorker<TestJobParameters, FirstStepOutput> last = (step, sink) -> {
+			fail("We should never hit this last step");
+			return RunOutcome.SUCCESS;
+		};
+		// job definition
+		String jobId = new Exception().getStackTrace()[0].getMethodName();
+		JobDefinition<? extends IModelJson> jd = JobDefinition.newBuilder()
+			.setJobDefinitionId(jobId)
+			.setJobDescription("test job")
+			.setJobDefinitionVersion(TEST_JOB_VERSION)
+			.setParametersType(TestJobParameters.class)
+			.gatedExecution()
+			.addFirstStep(
+				FIRST_STEP_ID,
+				"Test first step",
+				FirstStepOutput.class,
+				first
+			)
+			.addLastStep(
+				LAST_STEP_ID,
+				"Test last step",
+				last
+			)
+			.build();
+		myJobDefinitionRegistry.addJobDefinition(jd);
+		// test
+		JobInstanceStartRequest request = buildRequest(jobId);
+		myFirstStepLatch.setExpectedCount(1);
+		String instanceId = myJobCoordinator.startInstance(request);
+		JobInstance instance = myBatch2JobHelper.awaitJobHitsStatusInTime(instanceId,
+			12, // we want to wait a long time (2 min here) cause backoff is incremental
+			StatusEnum.FAILED, StatusEnum.ERRORED // error states
+		);
+
+		assertEquals(MAX_CHUNK_ERROR_COUNT + 1, counter.get());
+		assertEquals(MAX_CHUNK_ERROR_COUNT, interceptorCounter.get());
+
+		assertTrue(instance.getStatus() == StatusEnum.FAILED
+			|| instance.getStatus() == StatusEnum.ERRORED);
+	}
+
 	@Nonnull
 	private JobInstanceStartRequest buildRequest(String jobId) {
 		JobInstanceStartRequest request = new JobInstanceStartRequest();

hapi-fhir-jpaserver-test-utilities/src/test/java/ca/uhn/fhir/jpa/bulk/imprt2/BulkImportR4Test.java

@@ -185,7 +185,7 @@ public class BulkImportR4Test extends BaseJpaR4Test {
 				assertNotNull(instance.getCreateTime());
 				assertNotNull(instance.getStartTime());
 				assertNull(instance.getEndTime());
-				assertThat(instance.getErrorMessage(), containsString("NullPointerException: This is an exception"));
+				assertThat(instance.getErrorMessage(), containsString("This is an exception"));
 			});
 
 		} finally {

hapi-fhir-jpaserver-test-utilities/src/test/java/ca/uhn/fhir/jpa/delete/job/ReindexJobTest.java

@@ -129,7 +129,7 @@ public class ReindexJobTest extends BaseJpaR4Test {
 		// Verify
 
 		assertEquals(StatusEnum.ERRORED, outcome.getStatus());
-		assertEquals("java.lang.NullPointerException: foo message", outcome.getErrorMessage());
+		assertEquals("foo message", outcome.getErrorMessage());
 	}
 
 	@Test

hapi-fhir-jpaserver-uhnfhirtest/pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-fhir</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

hapi-fhir-server-mdm/pom.xml

@@ -7,7 +7,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
 

hapi-fhir-server-openapi/pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
 

hapi-fhir-server/pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
 

hapi-fhir-spring-boot/hapi-fhir-spring-boot-autoconfigure/pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
 

hapi-fhir-spring-boot/hapi-fhir-spring-boot-samples/hapi-fhir-spring-boot-sample-client-apache/pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-fhir-spring-boot-samples</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 	</parent>
 
 	<artifactId>hapi-fhir-spring-boot-sample-client-apache</artifactId>

hapi-fhir-spring-boot/hapi-fhir-spring-boot-samples/hapi-fhir-spring-boot-sample-client-okhttp/pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-fhir-spring-boot-samples</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 	</parent>
 
 	<artifactId>hapi-fhir-spring-boot-sample-client-okhttp</artifactId>

hapi-fhir-spring-boot/hapi-fhir-spring-boot-samples/hapi-fhir-spring-boot-sample-server-jersey/pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-fhir-spring-boot-samples</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 	</parent>
 
 	<artifactId>hapi-fhir-spring-boot-sample-server-jersey</artifactId>

hapi-fhir-spring-boot/hapi-fhir-spring-boot-samples/pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-fhir-spring-boot</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 	</parent>
 
 	<artifactId>hapi-fhir-spring-boot-samples</artifactId>

hapi-fhir-spring-boot/hapi-fhir-spring-boot-starter/pom.xml

@@ -5,7 +5,7 @@
    <parent>
       <groupId>ca.uhn.hapi.fhir</groupId>
       <artifactId>hapi-deployable-pom</artifactId>
-      <version>6.1.0-PRE13-SNAPSHOT</version>
+      <version>6.1.0-PRE14-SNAPSHOT</version>
       <relativePath>../../hapi-deployable-pom/pom.xml</relativePath>
    </parent>
 

hapi-fhir-spring-boot/pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-fhir</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

hapi-fhir-sql-migrate/pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
 

hapi-fhir-storage-batch2-jobs/pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
 	<modelVersion>4.0.0</modelVersion>

hapi-fhir-storage-batch2-jobs/src/main/java/ca/uhn/fhir/batch2/jobs/reindex/ReindexJobParametersValidator.java

@@ -21,13 +21,19 @@ package ca.uhn.fhir.batch2.jobs.reindex;
  */
 
 import ca.uhn.fhir.batch2.api.IJobParametersValidator;
+import ca.uhn.fhir.batch2.jobs.parameters.PartitionedUrl;
 import ca.uhn.fhir.batch2.jobs.parameters.UrlListValidator;
+import ca.uhn.fhir.util.UrlUtil;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 
+import java.util.ArrayList;
 import java.util.List;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 public class ReindexJobParametersValidator implements IJobParametersValidator<ReindexJobParameters> {
+
 	private final UrlListValidator myUrlListValidator;
 
 	public ReindexJobParametersValidator(UrlListValidator theUrlListValidator) {
@@ -37,6 +43,21 @@ public class ReindexJobParametersValidator implements IJobParametersValidator<Re
 	@Nullable
 	@Override
 	public List<String> validate(@NotNull ReindexJobParameters theParameters) {
-		return myUrlListValidator.validatePartitionedUrls(theParameters.getPartitionedUrls());
+		List<String> errors = myUrlListValidator.validatePartitionedUrls(theParameters.getPartitionedUrls());
+
+		if (errors == null || errors.isEmpty()) {
+			// only check if there's no other errors (new list to fix immutable issues)
+			errors = new ArrayList<>();
+			List<PartitionedUrl> urls = theParameters.getPartitionedUrls();
+			for (PartitionedUrl purl : urls) {
+				String url = purl.getUrl();
+
+				if (url.contains(" ") || url.contains("\n") || url.contains("\t")) {
+					errors.add("Invalid URL. URL cannot contain spaces : " + url);
+				}
+			}
+		}
+
+		return errors;
 	}
 }

hapi-fhir-storage-batch2-jobs/src/test/java/ca/uhn/fhir/batch2/jobs/reindex/ReindexJobParametersValidatorTest.java

@@ -0,0 +1,47 @@
+package ca.uhn.fhir.batch2.jobs.reindex;
+
+import ca.uhn.fhir.batch2.jobs.parameters.UrlListValidator;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.beans.factory.annotation.Value;
+
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@ExtendWith(MockitoExtension.class)
+public class ReindexJobParametersValidatorTest {
+
+	@Mock
+	private UrlListValidator myListValidator;
+
+	@InjectMocks
+	private ReindexJobParametersValidator myValidator;
+
+	@ParameterizedTest
+	@ValueSource(strings = { "\n", " ", "\t" })
+	public void validate_urlWithSpace_fails(String theWhiteSpaceChar) {
+		List<String> errors = runTestWithUrl("Patient," + theWhiteSpaceChar + "Practitioner");
+
+		// verify
+		assertFalse(errors.isEmpty());
+		assertTrue(errors.get(0).contains("Invalid URL. URL cannot contain spaces"));
+	}
+
+	private List<String> runTestWithUrl(String theUrl) {
+		// setup
+		ReindexJobParameters parameters = new ReindexJobParameters();
+		parameters.addUrl(theUrl);
+
+		// test
+		List<String> errors = myValidator.validate(parameters);
+
+		return errors;
+	}
+}

hapi-fhir-storage-batch2/pom.xml

@@ -6,7 +6,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
 

hapi-fhir-storage-batch2/src/main/java/ca/uhn/fhir/batch2/api/IJobPersistence.java

@@ -22,6 +22,7 @@ package ca.uhn.fhir.batch2.api;
 
 import ca.uhn.fhir.batch2.coordinator.BatchWorkChunk;
 import ca.uhn.fhir.batch2.model.JobInstance;
+import ca.uhn.fhir.batch2.model.MarkWorkChunkAsErrorRequest;
 import ca.uhn.fhir.batch2.model.StatusEnum;
 import ca.uhn.fhir.batch2.model.WorkChunk;
 
@@ -100,8 +101,26 @@ public interface IJobPersistence {
 	 *
 	 * @param theChunkId The chunk ID
 	 */
+	@Deprecated
 	void markWorkChunkAsErroredAndIncrementErrorCount(String theChunkId, String theErrorMessage);
 
+	/**
+	 * Marks a given chunk as having errored (ie, may be recoverable)
+	 *
+	 * Returns the work chunk.
+	 *
+	 * NB: For backwards compatibility reasons, it could be an empty optional, but
+	 * this doesn't mean it has no workchunk (just implementers are not updated)
+	 *
+	 * @param theParameters - the parameters for marking the workchunk with error
+	 * @return - workchunk optional, if available.
+	 */
+	default Optional<WorkChunk> markWorkChunkAsErroredAndIncrementErrorCount(MarkWorkChunkAsErrorRequest theParameters) {
+		// old method - please override me
+		markWorkChunkAsErroredAndIncrementErrorCount(theParameters.getChunkId(), theParameters.getErrorMsg());
+		return Optional.empty(); // returning empty so as not to break implementers
+	}
+
 	/**
 	 * Marks a given chunk as having failed (i.e. probably not recoverable)
 	 *

hapi-fhir-storage-batch2/src/main/java/ca/uhn/fhir/batch2/coordinator/StepExecutionSvc.java

@@ -36,6 +36,7 @@ import ca.uhn.fhir.batch2.model.JobDefinition;
 import ca.uhn.fhir.batch2.model.JobDefinitionStep;
 import ca.uhn.fhir.batch2.model.JobInstance;
 import ca.uhn.fhir.batch2.model.JobWorkCursor;
+import ca.uhn.fhir.batch2.model.MarkWorkChunkAsErrorRequest;
 import ca.uhn.fhir.batch2.model.StatusEnum;
 import ca.uhn.fhir.batch2.model.WorkChunk;
 import ca.uhn.fhir.i18n.Msg;
@@ -48,10 +49,23 @@ import javax.annotation.Nullable;
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Optional;
 
 public class StepExecutionSvc {
 	private static final Logger ourLog = LoggerFactory.getLogger(StepExecutionSvc.class);
 
+	// TODO
+	/**
+	 * This retry only works if your channel producer supports
+	 * retries on message processing exceptions.
+	 *
+	 * What's more, we may one day want to have this configurable
+	 * by the caller.
+	 * But since this is not a feature of HAPI,
+	 * this has not been done yet.
+	 */
+	public static final int MAX_CHUNK_ERROR_COUNT = 3;
+
 	private final IJobPersistence myJobPersistence;
 	private final BatchJobSender myBatchJobSender;
 
@@ -303,7 +317,20 @@ public class StepExecutionSvc {
 		} catch (Exception e) {
 			ourLog.error("Failure executing job {} step {}", jobDefinitionId, targetStepId, e);
 			if (theStepExecutionDetails.hasAssociatedWorkChunk()) {
-				myJobPersistence.markWorkChunkAsErroredAndIncrementErrorCount(chunkId, e.toString());
+				MarkWorkChunkAsErrorRequest parameters = new MarkWorkChunkAsErrorRequest();
+				parameters.setChunkId(chunkId);
+				parameters.setErrorMsg(e.getMessage());
+				Optional<WorkChunk> updatedOp = myJobPersistence.markWorkChunkAsErroredAndIncrementErrorCount(parameters);
+				if (updatedOp.isPresent()) {
+					WorkChunk chunk = updatedOp.get();
+
+					// TODO - marking for posterity
+					// see comments on MAX_CHUNK_ERROR_COUNT
+					if (chunk.getErrorCount() > MAX_CHUNK_ERROR_COUNT) {
+						myJobPersistence.markWorkChunkAsFailed(chunkId, "Too many errors: " + chunk.getErrorCount());
+						return false;
+					}
+				}
 			}
 			throw new JobStepFailedException(Msg.code(2041) + e.getMessage(), e);
 		} catch (Throwable t) {

hapi-fhir-storage-batch2/src/main/java/ca/uhn/fhir/batch2/coordinator/SynchronizedJobPersistenceWrapper.java

@@ -23,6 +23,7 @@ package ca.uhn.fhir.batch2.coordinator;
 import ca.uhn.fhir.batch2.api.IJobPersistence;
 import ca.uhn.fhir.batch2.api.JobOperationResultJson;
 import ca.uhn.fhir.batch2.model.JobInstance;
+import ca.uhn.fhir.batch2.model.MarkWorkChunkAsErrorRequest;
 import ca.uhn.fhir.batch2.model.StatusEnum;
 import ca.uhn.fhir.batch2.model.WorkChunk;
 
@@ -92,6 +93,11 @@ public class SynchronizedJobPersistenceWrapper implements IJobPersistence {
 		myWrap.markWorkChunkAsErroredAndIncrementErrorCount(theChunkId, theErrorMessage);
 	}
 
+	@Override
+	public Optional<WorkChunk> markWorkChunkAsErroredAndIncrementErrorCount(MarkWorkChunkAsErrorRequest theParameters) {
+		return myWrap.markWorkChunkAsErroredAndIncrementErrorCount(theParameters);
+	}
+
 	@Override
 	public synchronized void markWorkChunkAsFailed(String theChunkId, String theErrorMessage) {
 		myWrap.markWorkChunkAsFailed(theChunkId, theErrorMessage);

hapi-fhir-storage-batch2/src/main/java/ca/uhn/fhir/batch2/model/MarkWorkChunkAsErrorRequest.java

@@ -0,0 +1,33 @@
+package ca.uhn.fhir.batch2.model;
+
+public class MarkWorkChunkAsErrorRequest {
+	private String myChunkId;
+
+	private String myErrorMsg;
+
+	private boolean myIncludeData;
+
+	public String getChunkId() {
+		return myChunkId;
+	}
+
+	public void setChunkId(String theChunkId) {
+		myChunkId = theChunkId;
+	}
+
+	public String getErrorMsg() {
+		return myErrorMsg;
+	}
+
+	public void setErrorMsg(String theErrorMsg) {
+		myErrorMsg = theErrorMsg;
+	}
+
+	public boolean isIncludeData() {
+		return myIncludeData;
+	}
+
+	public void setIncludeData(boolean theIncludeData) {
+		myIncludeData = theIncludeData;
+	}
+}

hapi-fhir-storage-batch2/src/main/java/ca/uhn/fhir/batch2/model/StatusEnum.java

@@ -116,6 +116,7 @@ public enum StatusEnum {
 		retVal = ourNotEndedStatuses;
 		return retVal;
 	}
+
 	@Nonnull
 	private static void initializeStaticEndedStatuses() {
 		EnumSet<StatusEnum> endedSet = EnumSet.noneOf(StatusEnum.class);

hapi-fhir-storage-batch2/src/test/java/ca/uhn/fhir/batch2/coordinator/JobCoordinatorImplTest.java

@@ -12,6 +12,7 @@ import ca.uhn.fhir.batch2.model.JobInstance;
 import ca.uhn.fhir.batch2.model.JobInstanceStartRequest;
 import ca.uhn.fhir.batch2.model.JobWorkNotification;
 import ca.uhn.fhir.batch2.model.JobWorkNotificationJsonMessage;
+import ca.uhn.fhir.batch2.model.MarkWorkChunkAsErrorRequest;
 import ca.uhn.fhir.batch2.model.StatusEnum;
 import ca.uhn.fhir.batch2.model.WorkChunk;
 import ca.uhn.fhir.jpa.subscription.channel.api.IChannelReceiver;
@@ -245,8 +246,11 @@ public class JobCoordinatorImplTest extends BaseBatch2Test {
 		assertEquals(PARAM_2_VALUE, params.getParam2());
 		assertEquals(PASSWORD_VALUE, params.getPassword());
 
-		verify(myJobInstancePersister, times(1)).markWorkChunkAsErroredAndIncrementErrorCount(eq(CHUNK_ID), myErrorMessageCaptor.capture());
+		ArgumentCaptor<MarkWorkChunkAsErrorRequest> parametersArgumentCaptor = ArgumentCaptor.forClass(MarkWorkChunkAsErrorRequest.class);
-		assertEquals("java.lang.NullPointerException: This is an error message", myErrorMessageCaptor.getValue());
+		verify(myJobInstancePersister, times(1)).markWorkChunkAsErroredAndIncrementErrorCount(parametersArgumentCaptor.capture());
+		MarkWorkChunkAsErrorRequest capturedParams = parametersArgumentCaptor.getValue();
+		assertEquals(CHUNK_ID, capturedParams.getChunkId());
+		assertEquals("This is an error message", capturedParams.getErrorMsg());
 	}
 
 	@Test

hapi-fhir-storage-batch2/src/test/java/ca/uhn/fhir/batch2/coordinator/StepExecutionSvcTest.java

@@ -19,6 +19,7 @@ import ca.uhn.fhir.batch2.model.JobDefinitionReductionStep;
 import ca.uhn.fhir.batch2.model.JobDefinitionStep;
 import ca.uhn.fhir.batch2.model.JobInstance;
 import ca.uhn.fhir.batch2.model.JobWorkCursor;
+import ca.uhn.fhir.batch2.model.MarkWorkChunkAsErrorRequest;
 import ca.uhn.fhir.batch2.model.StatusEnum;
 import ca.uhn.fhir.batch2.model.WorkChunk;
 import ca.uhn.fhir.batch2.model.WorkChunkData;
@@ -34,9 +35,12 @@ import org.mockito.junit.jupiter.MockitoExtension;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
 import static org.mockito.ArgumentMatchers.any;
@@ -492,11 +496,70 @@ public class StepExecutionSvcTest {
 			fail("Expected Exception to be thrown");
 		} catch (JobStepFailedException jobStepFailedException) {
 			assertTrue(jobStepFailedException.getMessage().contains(msg));
-		} catch (Exception anythingElse) {
-			fail(anythingElse.getMessage());
 		}
 	}
 
+	@Test
+	public void doExecution_stepWorkerThrowsRandomExceptionForever_eventuallyMarksAsFailedAndReturnsFalse() {
+		// setup
+		int counter = 0;
+		AtomicInteger errorCounter = new AtomicInteger();
+		String errorMsg = "my Error Message";
+		JobInstance jobInstance = getTestJobInstance();
+		WorkChunk chunk = new WorkChunk();
+		chunk.setId("chunkId");
+		chunk.setData(new StepInputData());
+
+		JobWorkCursor<TestJobParameters, StepInputData, StepOutputData> workCursor = mock(JobWorkCursor.class);
+
+		JobDefinitionStep<TestJobParameters, StepInputData, StepOutputData> step = mockOutWorkCursor(StepType.INTERMEDIATE, workCursor, true, false);
+
+		// when
+		when(myNonReductionStep.run(any(), any()))
+			.thenThrow(new RuntimeException(errorMsg));
+		when(myJobPersistence.markWorkChunkAsErroredAndIncrementErrorCount(any(MarkWorkChunkAsErrorRequest.class)))
+			.thenAnswer((p) -> {
+				WorkChunk ec = new WorkChunk();
+				ec.setId(chunk.getId());
+				int count = errorCounter.getAndIncrement();
+				ec.setErrorCount(count);
+				return Optional.of(ec);
+			});
+
+		// test
+		Boolean processedOutcomeSuccessfully = null;
+		do {
+			try {
+				JobStepExecutorOutput<?, ?, ?> output = myExecutorSvc.doExecution(
+					workCursor,
+					jobInstance,
+					chunk
+				);
+				/*
+				 * Getting a value here means we are no longer
+				 * throwing exceptions. Which is the desired outcome.
+				 * We just now need to ensure that this outcome is
+				 * "false"
+				 */
+				processedOutcomeSuccessfully = output.isSuccessful();
+			} catch (JobStepFailedException ex) {
+				assertTrue(ex.getMessage().contains(errorMsg));
+				counter++;
+			}
+			/*
+			 * +2 because...
+			 * we check for > MAX_CHUNK_ERROR_COUNT (+1)
+			 * we want it to run one extra time here (+1)
+			 */
+		} while (processedOutcomeSuccessfully == null && counter < StepExecutionSvc.MAX_CHUNK_ERROR_COUNT + 2);
+
+		// verify
+		assertNotNull(processedOutcomeSuccessfully);
+		// +1 because of the > MAX_CHUNK_ERROR_COUNT check
+		assertEquals(StepExecutionSvc.MAX_CHUNK_ERROR_COUNT  + 1, counter);
+		assertFalse(processedOutcomeSuccessfully);
+	}
+
 	private void runExceptionThrowingTest(Exception theExceptionToThrow) {
 		// setup
 		JobInstance jobInstance = getTestJobInstance();

hapi-fhir-storage-mdm/pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
 	<modelVersion>4.0.0</modelVersion>

hapi-fhir-storage-test-utilities/pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
 	<modelVersion>4.0.0</modelVersion>

hapi-fhir-storage/pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
 

hapi-fhir-storage/src/main/java/ca/uhn/fhir/jpa/patch/XmlPatchUtils.java

@@ -1,7 +1,7 @@
 package ca.uhn.fhir.jpa.patch;
 
-import ca.uhn.fhir.i18n.Msg;
 import ca.uhn.fhir.context.FhirContext;
+import ca.uhn.fhir.i18n.Msg;
 import ca.uhn.fhir.rest.api.Constants;
 import ca.uhn.fhir.rest.server.exceptions.InternalErrorException;
 import com.github.dnault.xmlpatch.Patcher;

hapi-fhir-storage/src/main/java/ca/uhn/fhir/jpa/subscription/channel/impl/LinkedBlockingChannel.java

@@ -24,6 +24,7 @@ import ca.uhn.fhir.jpa.subscription.channel.api.IChannelProducer;
 import ca.uhn.fhir.jpa.subscription.channel.api.IChannelReceiver;
 import org.springframework.messaging.support.ExecutorSubscribableChannel;
 
+import java.util.ArrayList;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadPoolExecutor;
@@ -44,9 +45,7 @@ public class LinkedBlockingChannel extends ExecutorSubscribableChannel implement
 	}
 
 	public void clearInterceptorsForUnitTest() {
-		while (getInterceptors().size() > 0) {
+		setInterceptors(new ArrayList<>());
-			removeInterceptor(0);
-		}
 	}
 
 	@Override

hapi-fhir-storage/src/main/java/ca/uhn/fhir/jpa/subscription/channel/impl/LinkedBlockingChannelFactory.java

@@ -34,6 +34,7 @@ import org.apache.commons.lang3.concurrent.BasicThreadFactory;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.annotation.Nonnull;
 import javax.annotation.PreDestroy;
 import java.util.Collections;
 import java.util.HashMap;
@@ -76,8 +77,11 @@ public class LinkedBlockingChannelFactory implements IChannelFactory {
 		// TODO - does this need retry settings?
 		final String channelName = myChannelNamer.getChannelName(theChannelName, theChannelSettings);
 
-		return myChannels.computeIfAbsent(channelName, t -> {
+		return myChannels.computeIfAbsent(channelName, t -> buildLinkedBlockingChannel(theConcurrentConsumers, channelName));
+	}
 
+	@Nonnull
+	private LinkedBlockingChannel buildLinkedBlockingChannel(int theConcurrentConsumers, String channelName) {
 		String threadNamingPattern = channelName + "-%d";
 
 		ThreadFactory threadFactory = new BasicThreadFactory.Builder()
@@ -108,9 +112,9 @@ public class LinkedBlockingChannelFactory implements IChannelFactory {
 			queue,
 			threadFactory,
 			rejectedExecutionHandler);
-			return new LinkedBlockingChannel(channelName, executor, queue);
 
-		});
+		LinkedBlockingChannel retval = new LinkedBlockingChannel(channelName, executor, queue);
+		return retval;
 	}
 
 	private Thread.UncaughtExceptionHandler uncaughtExceptionHandler(String theChannelName) {

hapi-fhir-structures-dstu2.1/pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
 

hapi-fhir-structures-dstu2/pom.xml

@@ -4,7 +4,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
 

hapi-fhir-structures-dstu3/pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
 

hapi-fhir-structures-hl7org-dstu2/pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
 

hapi-fhir-structures-r4/pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
 

hapi-fhir-structures-r5/pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
 

hapi-fhir-test-utilities/pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
 

hapi-fhir-testpage-overlay/pom.xml

@@ -4,7 +4,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-fhir</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

hapi-fhir-validation-resources-dstu2.1/pom.xml

@@ -4,7 +4,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
 

hapi-fhir-validation-resources-dstu2/pom.xml

@@ -4,7 +4,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
 

hapi-fhir-validation-resources-dstu3/pom.xml

@@ -4,7 +4,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
 

hapi-fhir-validation-resources-r4/pom.xml

@@ -4,7 +4,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
 

hapi-fhir-validation-resources-r5/pom.xml

@@ -4,7 +4,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
 

hapi-fhir-validation/pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-deployable-pom</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../hapi-deployable-pom/pom.xml</relativePath>
 	</parent>
 

hapi-tinder-plugin/pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-fhir</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 
@@ -58,37 +58,37 @@
 		<dependency>
 			<groupId>ca.uhn.hapi.fhir</groupId>
 			<artifactId>hapi-fhir-structures-dstu3</artifactId>
-			<version>6.1.0-PRE13-SNAPSHOT</version>
+			<version>6.1.0-PRE14-SNAPSHOT</version>
 		</dependency>
 		<dependency>
 			<groupId>ca.uhn.hapi.fhir</groupId>
 			<artifactId>hapi-fhir-structures-hl7org-dstu2</artifactId>
-			<version>6.1.0-PRE13-SNAPSHOT</version>
+			<version>6.1.0-PRE14-SNAPSHOT</version>
 		</dependency>
 		<dependency>
 			<groupId>ca.uhn.hapi.fhir</groupId>
 			<artifactId>hapi-fhir-structures-r4</artifactId>
-			<version>6.1.0-PRE13-SNAPSHOT</version>
+			<version>6.1.0-PRE14-SNAPSHOT</version>
 		</dependency>
 		<dependency>
 			<groupId>ca.uhn.hapi.fhir</groupId>
 			<artifactId>hapi-fhir-structures-r5</artifactId>
-			<version>6.1.0-PRE13-SNAPSHOT</version>
+			<version>6.1.0-PRE14-SNAPSHOT</version>
 		</dependency>
 		<dependency>
 			<groupId>ca.uhn.hapi.fhir</groupId>
 			<artifactId>hapi-fhir-validation-resources-dstu2</artifactId>
-			<version>6.1.0-PRE13-SNAPSHOT</version>
+			<version>6.1.0-PRE14-SNAPSHOT</version>
 		</dependency>
 		<dependency>
 			<groupId>ca.uhn.hapi.fhir</groupId>
 			<artifactId>hapi-fhir-validation-resources-dstu3</artifactId>
-			<version>6.1.0-PRE13-SNAPSHOT</version>
+			<version>6.1.0-PRE14-SNAPSHOT</version>
 		</dependency>
 		<dependency>
 			<groupId>ca.uhn.hapi.fhir</groupId>
 			<artifactId>hapi-fhir-validation-resources-r4</artifactId>
-			<version>6.1.0-PRE13-SNAPSHOT</version>
+			<version>6.1.0-PRE14-SNAPSHOT</version>
 		</dependency>
 		<dependency>
 			<groupId>org.apache.velocity</groupId>

hapi-tinder-test/pom.xml

@@ -4,7 +4,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-fhir</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

pom.xml

@@ -6,7 +6,7 @@
 	<groupId>ca.uhn.hapi.fhir</groupId>
 	<artifactId>hapi-fhir</artifactId>
 	<packaging>pom</packaging>
-	<version>6.1.0-PRE13-SNAPSHOT</version>
+	<version>6.1.0-PRE14-SNAPSHOT</version>
 	<name>HAPI-FHIR</name>
 	<description>An open-source implementation of the FHIR specification in Java.</description>
 	<url>https://hapifhir.io</url>
@@ -2015,7 +2015,7 @@
                             <groupId>ca.uhn.hapi.fhir</groupId>
                             <artifactId>hapi-fhir-checkstyle</artifactId>
 									<!-- Remember to bump this when you upgrade the version -->
-                            <version>6.1.0-PRE13-SNAPSHOT</version>
+                            <version>6.1.0-PRE14-SNAPSHOT</version>
                         </dependency>
 					</dependencies>
 				</plugin>

tests/hapi-fhir-base-test-jaxrsserver-kotlin/pom.xml

@@ -6,7 +6,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-fhir</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../../pom.xml</relativePath>
 	</parent>
 

tests/hapi-fhir-base-test-mindeps-client/pom.xml

@@ -4,7 +4,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-fhir</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../../pom.xml</relativePath>
 	</parent>
 

tests/hapi-fhir-base-test-mindeps-server/pom.xml

@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>ca.uhn.hapi.fhir</groupId>
 		<artifactId>hapi-fhir</artifactId>
-		<version>6.1.0-PRE13-SNAPSHOT</version>
+		<version>6.1.0-PRE14-SNAPSHOT</version>
 		<relativePath>../../pom.xml</relativePath>
 	</parent>