Fix metrics collection in FaultTolerantChunkProcessor

Before this commit, metrics were not collected in a fault-tolerant step.
This commit updates the FaultTolerantChunkProcessor to collect metrics.

For the record, chunk scanning is not covered for two reasons:

1. When scanning a chunk, there is a single item in each write operation,
so it would be incorrect to report a metric called "chunk.write" for a
single item. We could argue that it is a singleton chunk, but still..
If we want to time scanned (aka individual) items, we need a more fine
grained timer called "scanned.item.write" for example.

2. The end result can be confusing and might distort the overall metrics
view in case of errors (because of the noisy metrics of additional transactions
for individual items).

As a reminder, the goal of the "chunk.write" metric is to give an overview
of the write operation time of the whole chunk and not to time each item
individually (this could be done using an `ItemWriteListener` if needed).

Resolves #3664

Author: fmbenhassine

spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2006-2019 the original author or authors.
+ * Copyright 2006-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -23,11 +23,15 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
+import io.micrometer.core.instrument.Tag;
+import io.micrometer.core.instrument.Timer;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
 import org.springframework.batch.core.StepContribution;
+import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.listener.StepListenerFailedException;
+import org.springframework.batch.core.metrics.BatchMetrics;
 import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy;
 import org.springframework.batch.core.step.skip.NonSkippableProcessException;
 import org.springframework.batch.core.step.skip.SkipLimitExceededException;
@@ -222,6 +226,8 @@ protected Chunk<O> transform(final StepContribution contribution, Chunk<I> input
 
 				@Override
 				public O doWithRetry(RetryContext context) throws Exception {
+					Timer.Sample sample = BatchMetrics.createTimerSample();
+					String status = BatchMetrics.STATUS_SUCCESS;
 					O output = null;
 					try {
 						count.incrementAndGet();
@@ -239,6 +245,7 @@ public O doWithRetry(RetryContext context) throws Exception {
 						}
 					}
 					catch (Exception e) {
+						status = BatchMetrics.STATUS_FAILURE;
 						if (rollbackClassifier.classify(e)) {
 							// Default is to rollback unless the classifier
 							// allows us to continue
@@ -262,6 +269,9 @@ else if (shouldSkip(itemProcessSkipPolicy, e, contribution.getStepSkipCount()))
 									e);
 						}
 					}
+					finally {
+						stopTimer(sample, contribution.getStepExecution(), "item.process", status, "Item processing");
+					}
 					if (output == null) {
 						// No need to re-process filtered items
 						iterator.remove();
@@ -332,10 +342,13 @@ public Object doWithRetry(RetryContext context) throws Exception {
 
 				if (!data.scanning()) {
 					chunkMonitor.setChunkSize(inputs.size());
+					Timer.Sample sample = BatchMetrics.createTimerSample();
+					String status = BatchMetrics.STATUS_SUCCESS;
 					try {
 						doWrite(outputs.getItems());
 					}
 					catch (Exception e) {
+						status = BatchMetrics.STATUS_FAILURE;
 						if (rollbackClassifier.classify(e)) {
 							throw e;
 						}
@@ -348,6 +361,9 @@ public Object doWithRetry(RetryContext context) throws Exception {
 						throw new ForceRollbackForWriteSkipException(
 								"Force rollback on skippable exception so that skipped item can be located.", e);
 					}
+					finally {
+						stopTimer(sample, contribution.getStepExecution(), "chunk.write", status, "Chunk writing");
+					}
 					contribution.incrementWriteCount(outputs.size());
 				}
 				else {

spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkProcessor.java

@@ -340,7 +340,7 @@ protected Chunk<O> transform(StepContribution contribution, Chunk<I> inputs) thr
 		return outputs;
 	}
 
-	private void stopTimer(Timer.Sample sample, StepExecution stepExecution, String metricName, String status, String description) {
+	protected void stopTimer(Timer.Sample sample, StepExecution stepExecution, String metricName, String status, String description) {
 		sample.stop(BatchMetrics.createTimer(metricName, description + " duration",
 				Tag.of("job.name", stepExecution.getJobExecution().getJobInstance().getJobName()),
 				Tag.of("step.name", stepExecution.getStepName()),

spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessorTests.java

@@ -27,6 +27,8 @@
 import org.junit.Before;
 import org.junit.Test;
 import org.springframework.batch.core.JobExecution;
+import org.springframework.batch.core.JobInstance;
+import org.springframework.batch.core.JobParameters;
 import org.springframework.batch.core.StepContribution;
 import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.listener.ItemListenerSupport;
@@ -55,7 +57,7 @@ public class FaultTolerantChunkProcessorTests {
 	private FaultTolerantChunkProcessor<String, String> processor;
 
 	private StepContribution contribution = new StepExecution("foo",
-			new JobExecution(0L)).createStepContribution();
+			new JobExecution(new JobInstance(0L, "job"), new JobParameters())).createStepContribution();
 
 	@Before
 	public void setUp() {

spring-batch-samples/src/test/java/org/springframework/batch/sample/metrics/BatchMetricsTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2019-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -37,8 +37,6 @@
 import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
 import org.springframework.batch.core.launch.JobLauncher;
 import org.springframework.batch.core.metrics.BatchMetrics;
-import org.springframework.batch.item.ItemReader;
-import org.springframework.batch.item.ItemWriter;
 import org.springframework.batch.item.support.ListItemReader;
 import org.springframework.batch.repeat.RepeatStatus;
 import org.springframework.context.ApplicationContext;
@@ -51,9 +49,12 @@
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
+/**
+ * @author Mahmoud Ben Hassine
+ */
 public class BatchMetricsTests {
 
-	private static final int EXPECTED_SPRING_BATCH_METRICS = 6;
+	private static final int EXPECTED_SPRING_BATCH_METRICS = 10;
 
 	@Test
 	public void testCalculateDuration() {
@@ -147,6 +148,8 @@ public void testBatchMetrics() throws Exception {
 		List<Meter> meters = Metrics.globalRegistry.getMeters();
 		assertTrue(meters.size() >= EXPECTED_SPRING_BATCH_METRICS);
 
+		// Job metrics
+
 		try {
 			Metrics.globalRegistry.get("spring.batch.job")
 					.tag("name", "job")
@@ -164,6 +167,8 @@ public void testBatchMetrics() throws Exception {
 			fail("There should be a meter of type LONG_TASK_TIMER named spring.batch.job.active" +
 					" registered in the global registry: " + e.getMessage());
 		}
+		
+		// Step 1 (tasklet) metrics
 
 		try {
 			Metrics.globalRegistry.get("spring.batch.step")
@@ -175,6 +180,8 @@ public void testBatchMetrics() throws Exception {
 			fail("There should be a meter of type TIMER named spring.batch.step" +
 					" registered in the global registry: " + e.getMessage());
 		}
+		
+		// Step 2 (simple chunk-oriented) metrics
 
 		try {
 			Metrics.globalRegistry.get("spring.batch.step")
@@ -219,6 +226,52 @@ public void testBatchMetrics() throws Exception {
 			fail("There should be a meter of type TIMER named spring.batch.chunk.write" +
 					" registered in the global registry: " + e.getMessage());
 		}
+		
+		// Step 3 (fault-tolerant chunk-oriented) metrics
+
+		try {
+			Metrics.globalRegistry.get("spring.batch.step")
+					.tag("name", "step3")
+					.tag("job.name", "job")
+					.tag("status", "COMPLETED")
+					.timer();
+		} catch (Exception e) {
+			fail("There should be a meter of type TIMER named spring.batch.step" +
+					" registered in the global registry: " + e.getMessage());
+		}
+
+		try {
+			Metrics.globalRegistry.get("spring.batch.item.read")
+					.tag("job.name", "job")
+					.tag("step.name", "step3")
+					.tag("status", "SUCCESS")
+					.timer();
+		} catch (Exception e) {
+			fail("There should be a meter of type TIMER named spring.batch.item.read" +
+					" registered in the global registry: " + e.getMessage());
+		}
+
+		try {
+			Metrics.globalRegistry.get("spring.batch.item.process")
+					.tag("job.name", "job")
+					.tag("step.name", "step3")
+					.tag("status", "SUCCESS")
+					.timer();
+		} catch (Exception e) {
+			fail("There should be a meter of type TIMER named spring.batch.item.process" +
+					" registered in the global registry: " + e.getMessage());
+		}
+
+		try {
+			Metrics.globalRegistry.get("spring.batch.chunk.write")
+					.tag("job.name", "job")
+					.tag("step.name", "step3")
+					.tag("status", "SUCCESS")
+					.timer();
+		} catch (Exception e) {
+			fail("There should be a meter of type TIMER named spring.batch.chunk.write" +
+					" registered in the global registry: " + e.getMessage());
+		}
 	}
 
 	@Configuration
@@ -241,25 +294,23 @@ public Step step1() {
 		}
 
 		@Bean
-		public ItemReader<Integer> itemReader() {
-			return new ListItemReader<>(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10));
-		}
-
-		@Bean
-		public ItemWriter<Integer> itemWriter() {
-			return items -> {
-				for (Integer item : items) {
-					System.out.println("item = " + item);
-				}
-			};
+		public Step step2() {
+			return stepBuilderFactory.get("step2")
+					.<Integer, Integer>chunk(2)
+					.reader(new ListItemReader<>(Arrays.asList(1, 2, 3, 4, 5)))
+					.writer(items -> items.forEach(System.out::println))
+					.build();
 		}
 
 		@Bean
-		public Step step2() {
-			return stepBuilderFactory.get("step2")
-					.<Integer, Integer>chunk(5)
-					.reader(itemReader())
-					.writer(itemWriter())
+		public Step step3() {
+			return stepBuilderFactory.get("step3")
+					.<Integer, Integer>chunk(2)
+					.reader(new ListItemReader<>(Arrays.asList(6, 7, 8, 9, 10)))
+					.writer(items -> items.forEach(System.out::println))
+					.faultTolerant()
+					.skip(Exception.class)
+					.skipLimit(3)
 					.build();
 		}
 
@@ -268,6 +319,7 @@ public Job job() {
 			return jobBuilderFactory.get("job")
 					.start(step1())
 					.next(step2())
+					.next(step3())
 					.build();
 		}
 	}