Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix metrics collection in FaultTolerantChunkProcessor

Browse Source 
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
This commit is contained in:
Mahmoud Ben Hassine
2020-02-17 14:10:02 +01:00
parent 0a0a2ec802
commit 274cad09f3
4 changed files with 94 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<I, O> extends SimpleChunkProcessor<I, O
@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 class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
}
}
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 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
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 class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
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 class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
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 {

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

@@ -340,7 +340,7 @@ public class SimpleChunkProcessor<I, O> implements ChunkProcessor<I>, 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()),

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

@@ -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<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() {

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

@@ -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<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 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<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);
}
};
}
@Bean
public Step step2() {
return stepBuilderFactory.get("step2")
.<Integer, Integer>chunk(5)
.reader(itemReader())
.writer(itemWriter())
.<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 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 class BatchMetricsTests {
return jobBuilderFactory.get("job")
.start(step1())
.next(step2())
.next(step3())
.build();
}
}