diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java index b61b1174c..5d2ad0426 100755 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java +++ b/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.List; 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 @@ public class FaultTolerantChunkProcessor extends SimpleChunkProcessor extends SimpleChunkProcessor extends SimpleChunkProcessor extends SimpleChunkProcessor extends SimpleChunkProcessor implements ChunkProcessor, Initializi 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()), diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessorTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessorTests.java index 7f5502513..ec854c422 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessorTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessorTests.java @@ -27,6 +27,8 @@ import java.util.List; 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 processor; private StepContribution contribution = new StepExecution("foo", - new JobExecution(0L)).createStepContribution(); + new JobExecution(new JobInstance(0L, "job"), new JobParameters())).createStepContribution(); @Before public void setUp() { diff --git a/spring-batch-samples/src/test/java/org/springframework/batch/sample/metrics/BatchMetricsTests.java b/spring-batch-samples/src/test/java/org/springframework/batch/sample/metrics/BatchMetricsTests.java index e9fea17f6..efbdbcf50 100644 --- a/spring-batch-samples/src/test/java/org/springframework/batch/sample/metrics/BatchMetricsTests.java +++ b/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.JobBuilderFactory 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.assertNull; 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 class BatchMetricsTests { List 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 class BatchMetricsTests { 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 class BatchMetricsTests { 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 class BatchMetricsTests { 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 @@ -240,26 +293,24 @@ public class BatchMetricsTests { .build(); } - @Bean - public ItemReader itemReader() { - return new ListItemReader<>(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)); - } - - @Bean - public ItemWriter itemWriter() { - return items -> { - for (Integer item : items) { - System.out.println("item = " + item); - } - }; - } - @Bean public Step step2() { return stepBuilderFactory.get("step2") - .chunk(5) - .reader(itemReader()) - .writer(itemWriter()) + .chunk(2) + .reader(new ListItemReader<>(Arrays.asList(1, 2, 3, 4, 5))) + .writer(items -> items.forEach(System.out::println)) + .build(); + } + + @Bean + public Step step3() { + return stepBuilderFactory.get("step3") + .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 class BatchMetricsTests { return jobBuilderFactory.get("job") .start(step1()) .next(step2()) + .next(step3()) .build(); } }