MACRO NEPAL
Elastic APM Integration and Correlation in Java
awiskar acharya
November 17, 2025
10:57 am

Introduction

Elastic APM (Application Performance Monitoring) provides distributed tracing, performance monitoring, and error tracking for applications. Correlation in Elastic APM enables you to connect related events across different services and systems, providing end-to-end visibility into your application's performance and behavior.

This comprehensive guide covers Elastic APM integration, trace correlation, custom spans, and advanced monitoring patterns in Java applications.

Core Concepts

1. Elastic APM Components

  • APM Agent: Auto-instrumentation for Java applications
  • APM Server: Collects and processes monitoring data
  • Elasticsearch: Stores APM data
  • Kibana: Visualizes APM data

2. Correlation Concepts

  • Distributed Tracing: Track requests across service boundaries
  • Span Correlation: Connect related operations within a trace
  • Service Maps: Visualize service dependencies
  • Error Correlation: Link errors to specific traces and transactions

Project Setup and Dependencies

1. Maven Dependencies

<dependencies>
<!-- Elastic APM Agent (usually attached at runtime) -->
<!-- Spring Boot -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-jpa</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<!-- Database -->
<dependency>
<groupId>org.postgresql</groupId>
<artifactId>postgresql</artifactId>
<scope>runtime</scope>
</dependency>
<!-- HTTP Client -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-webflux</artifactId>
</dependency>
<!-- Elastic APM API -->
<dependency>
<groupId>co.elastic.apm</groupId>
<artifactId>apm-agent-api</artifactId>
<version>1.45.0</version>
</dependency>
<!-- Testing -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>

2. Application Configuration

# application.yml
elastic:
apm:
server-url: http://localhost:8200
service-name: order-service
environment: development
application-packages: com.example.elasticapm
spring:
datasource:
url: jdbc:postgresql://localhost:5432/orders
username: postgres
password: password
jpa:
hibernate:
ddl-auto: update
management:
endpoints:
web:
exposure:
include: health,metrics,info,prometheus
endpoint:
health:
show-details: always
logging:
level:
co.elastic.apm: DEBUG
com.example.elasticapm: INFO

Elastic APM Configuration

1. APM Agent Configuration

@Configuration
@Slf4j
public class ElasticApmConfig {
@Bean
@ConditionalOnMissingBean
public ApmAgentConfiguration apmAgentConfiguration(
@Value("${elastic.apm.server-url}") String serverUrl,
@Value("${elastic.apm.service-name}") String serviceName,
@Value("${elastic.apm.environment}") String environment,
@Value("${elastic.apm.application-packages}") String applicationPackages) {
return ApmAgentConfiguration.builder()
.serverUrl(serverUrl)
.serviceName(serviceName)
.environment(environment)
.applicationPackages(applicationPackages)
.build();
}
@Bean
public ElasticApmTracer elasticApmTracer(ApmAgentConfiguration config) {
// In a real scenario, this would be handled by the APM agent
// This bean demonstrates how you might configure custom tracing
log.info("Elastic APM configured for service: {} in environment: {}", 
config.getServiceName(), config.getEnvironment());
return new DefaultElasticApmTracer();
}
}
@Data
@Builder
public class ApmAgentConfiguration {
private String serverUrl;
private String serviceName;
private String environment;
private String applicationPackages;
private double transactionSampleRate = 1.0;
private int transactionMaxSpans = 500;
private long metricsInterval = 30L;
}

2. Custom Tracer Service

@Service
@Slf4j
public class ElasticApmTracerService {
private final co.elastic.apm.api.ElasticApm apm;
public ElasticApmTracerService() {
this.apm = co.elastic.apm.api.ElasticApm.getInstance();
}
/**
* Get current transaction
*/
public co.elastic.apm.api.Transaction getCurrentTransaction() {
return apm.currentTransaction();
}
/**
* Start a new custom transaction
*/
public co.elastic.apm.api.Transaction startTransaction(String name, String type, String subtype) {
co.elastic.apm.api.Transaction transaction = apm.startTransaction();
if (transaction != null) {
transaction.setName(name);
transaction.setType(type);
if (subtype != null) {
transaction.setSubtype(subtype);
}
}
return transaction;
}
/**
* Start a new span within current transaction
*/
public co.elastic.apm.api.Span startSpan(String name, String type, String subtype) {
co.elastic.apm.api.Span span = apm.currentSpan().startSpan();
if (span != null) {
span.setName(name);
span.setType(type);
if (subtype != null) {
span.setSubtype(subtype);
}
}
return span;
}
/**
* Add custom labels to current transaction
*/
public void addLabel(String key, String value) {
co.elastic.apm.api.Transaction transaction = getCurrentTransaction();
if (transaction != null) {
transaction.addLabel(key, value);
}
}
/**
* Add custom labels to current span
*/
public void addSpanLabel(String key, String value) {
co.elastic.apm.api.Span span = apm.currentSpan();
if (span != null) {
span.addLabel(key, value);
}
}
/**
* Capture exception in current transaction
*/
public void captureException(Throwable throwable) {
co.elastic.apm.api.Transaction transaction = getCurrentTransaction();
if (transaction != null) {
transaction.captureException(throwable);
}
}
/**
* Set transaction outcome
*/
public void setTransactionOutcome(boolean success) {
co.elastic.apm.api.Transaction transaction = getCurrentTransaction();
if (transaction != null) {
transaction.setOutcome(success ? "success" : "failure");
}
}
/**
* Set span outcome
*/
public void setSpanOutcome(boolean success) {
co.elastic.apm.api.Span span = apm.currentSpan();
if (span != null) {
span.setOutcome(success ? "success" : "failure");
}
}
/**
* Get current trace ID for correlation
*/
public String getCurrentTraceId() {
co.elastic.apm.api.Transaction transaction = getCurrentTransaction();
return transaction != null ? transaction.getTraceId() : null;
}
/**
* Get current transaction ID for correlation
*/
public String getCurrentTransactionId() {
co.elastic.apm.api.Transaction transaction = getCurrentTransaction();
return transaction != null ? transaction.getId() : null;
}
/**
* Inject trace context into headers for propagation
*/
public void injectTraceContext(Map<String, String> headers) {
co.elastic.apm.api.Transaction transaction = getCurrentTransaction();
if (transaction != null) {
transaction.injectTraceHeaders((carrier, key, value) -> {
if (carrier instanceof Map) {
((Map<String, String>) carrier).put(key, value);
}
});
}
}
/**
* Extract trace context from headers
*/
public co.elastic.apm.api.Transaction extractTraceContext(Map<String, String> headers) {
return apm.startTransactionWithRemoteParent(remoteParent -> {
headers.forEach(remoteParent::addTextHeader);
});
}
}

HTTP Correlation

1. HTTP Client with APM Correlation

@Component
@Slf4j
public class ApmCorrelatedHttpClient {
private final WebClient webClient;
private final ElasticApmTracerService tracerService;
private final ObjectMapper objectMapper;
public ApmCorrelatedHttpClient(WebClient.Builder webClientBuilder,
ElasticApmTracerService tracerService,
ObjectMapper objectMapper) {
this.tracerService = tracerService;
this.objectMapper = objectMapper;
this.webClient = webClientBuilder.build();
}
/**
* Make HTTP GET request with APM correlation
*/
public <T> Mono<T> getWithApmCorrelation(String url, Class<T> responseType) {
return Mono.fromCallable(() -> {
co.elastic.apm.api.Span span = tracerService.startSpan("HTTP GET", "external", "http");
try {
span.setName("GET " + url);
span.addLabel("http.url", url);
span.addLabel("http.method", "GET");
// Inject trace context into headers
Map<String, String> traceHeaders = new HashMap<>();
tracerService.injectTraceContext(traceHeaders);
// Make the HTTP call
String response = webClient.get()
.uri(url)
.headers(headers -> traceHeaders.forEach(headers::add))
.retrieve()
.bodyToMono(String.class)
.block();
span.addLabel("http.status_code", "200");
tracerService.setSpanOutcome(true);
return objectMapper.readValue(response, responseType);
} catch (Exception e) {
span.captureException(e);
tracerService.setSpanOutcome(false);
throw new HttpClientException("HTTP request failed: " + url, e);
} finally {
span.end();
}
});
}
/**
* Make HTTP POST request with APM correlation
*/
public <T, R> Mono<T> postWithApmCorrelation(String url, R body, Class<T> responseType) {
return Mono.fromCallable(() -> {
co.elastic.apm.api.Span span = tracerService.startSpan("HTTP POST", "external", "http");
try {
span.setName("POST " + url);
span.addLabel("http.url", url);
span.addLabel("http.method", "POST");
// Inject trace context
Map<String, String> traceHeaders = new HashMap<>();
tracerService.injectTraceContext(traceHeaders);
String response = webClient.post()
.uri(url)
.headers(headers -> traceHeaders.forEach(headers::add))
.bodyValue(body)
.retrieve()
.bodyToMono(String.class)
.block();
span.addLabel("http.status_code", "200");
tracerService.setSpanOutcome(true);
return objectMapper.readValue(response, responseType);
} catch (Exception e) {
span.captureException(e);
tracerService.setSpanOutcome(false);
throw new HttpClientException("HTTP POST request failed: " + url, e);
} finally {
span.end();
}
});
}
/**
* Make correlated call to another service
*/
public <T> Mono<T> callServiceWithCorrelation(String serviceName, String endpoint, 
Object request, Class<T> responseType) {
String url = buildServiceUrl(serviceName, endpoint);
co.elastic.apm.api.Span span = tracerService.startSpan("Service Call", "external", "http");
span.addLabel("service.name", serviceName);
span.addLabel("service.endpoint", endpoint);
try {
return postWithApmCorrelation(url, request, responseType)
.doOnSuccess(response -> {
span.addLabel("service.response", "success");
tracerService.setSpanOutcome(true);
})
.doOnError(error -> {
span.captureException(error);
span.addLabel("service.response", "error");
tracerService.setSpanOutcome(false);
});
} finally {
span.end();
}
}
private String buildServiceUrl(String serviceName, String endpoint) {
// In real implementation, use service discovery
Map<String, String> serviceUrls = Map.of(
"inventory-service", "http://inventory-service:8080",
"payment-service", "http://payment-service:8080",
"shipping-service", "http://shipping-service:8080"
);
String baseUrl = serviceUrls.get(serviceName);
if (baseUrl == null) {
throw new IllegalArgumentException("Unknown service: " + serviceName);
}
return baseUrl + endpoint;
}
}

2. HTTP Server Correlation Filter

@Component
@Slf4j
public class ApmCorrelationFilter implements Filter {
private final ElasticApmTracerService tracerService;
public ApmCorrelationFilter(ElasticApmTracerService tracerService) {
this.tracerService = tracerService;
}
@Override
public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain)
throws IOException, ServletException {
HttpServletRequest httpRequest = (HttpServletRequest) request;
HttpServletResponse httpResponse = (HttpServletResponse) response;
// Extract trace context from incoming headers
co.elastic.apm.api.Transaction transaction = extractTransactionFromHeaders(httpRequest);
if (transaction == null) {
// Start new transaction if no context found
transaction = tracerService.startTransaction(
httpRequest.getMethod() + " " + httpRequest.getRequestURI(),
"request",
null
);
}
try {
// Set transaction context
configureTransaction(transaction, httpRequest);
// Continue with the filter chain
chain.doFilter(request, response);
// Set transaction outcome based on response
setTransactionOutcome(transaction, httpResponse);
} catch (Exception e) {
transaction.captureException(e);
tracerService.setTransactionOutcome(false);
throw e;
} finally {
transaction.end();
}
}
private co.elastic.apm.api.Transaction extractTransactionFromHeaders(HttpServletRequest request) {
Map<String, String> headers = new HashMap<>();
Enumeration<String> headerNames = request.getHeaderNames();
while (headerNames.hasMoreElements()) {
String headerName = headerNames.nextElement();
headers.put(headerName, request.getHeader(headerName));
}
return tracerService.extractTraceContext(headers);
}
private void configureTransaction(co.elastic.apm.api.Transaction transaction, HttpServletRequest request) {
transaction.setName(request.getMethod() + " " + request.getRequestURI());
// Add request context as labels
transaction.addLabel("http.method", request.getMethod());
transaction.addLabel("http.url", request.getRequestURL().toString());
transaction.addLabel("http.query_string", request.getQueryString());
transaction.addLabel("http.user_agent", request.getHeader("User-Agent"));
transaction.addLabel("http.client_ip", getClientIp(request));
transaction.addLabel("service.instance.id", getInstanceId());
// Add business context if available
addBusinessContext(transaction, request);
}
private void setTransactionOutcome(co.elastic.apm.api.Transaction transaction, HttpServletResponse response) {
int status = response.getStatus();
transaction.addLabel("http.status_code", String.valueOf(status));
if (status >= 400) {
tracerService.setTransactionOutcome(false);
} else {
tracerService.setTransactionOutcome(true);
}
}
private String getClientIp(HttpServletRequest request) {
String xForwardedFor = request.getHeader("X-Forwarded-For");
if (xForwardedFor != null && !xForwardedFor.isEmpty()) {
return xForwardedFor.split(",")[0].trim();
}
return request.getRemoteAddr();
}
private String getInstanceId() {
try {
return InetAddress.getLocalHost().getHostName();
} catch (Exception e) {
return "unknown";
}
}
private void addBusinessContext(co.elastic.apm.api.Transaction transaction, HttpServletRequest request) {
// Extract business context from headers or request attributes
String userId = request.getHeader("X-User-ID");
String tenantId = request.getHeader("X-Tenant-ID");
String correlationId = request.getHeader("X-Correlation-ID");
if (userId != null) {
transaction.addLabel("user.id", userId);
}
if (tenantId != null) {
transaction.addLabel("tenant.id", tenantId);
}
if (correlationId != null) {
transaction.addLabel("correlation.id", correlationId);
}
}
}
// Register the filter
@Configuration
public class FilterConfig {
@Bean
public FilterRegistrationBean<ApmCorrelationFilter> apmCorrelationFilter(ElasticApmTracerService tracerService) {
FilterRegistrationBean<ApmCorrelationFilter> registrationBean = new FilterRegistrationBean<>();
registrationBean.setFilter(new ApmCorrelationFilter(tracerService));
registrationBean.addUrlPatterns("/*");
registrationBean.setOrder(1);
return registrationBean;
}
}

Database Correlation

1. JPA/Hibernate Correlation

@Component
@Slf4j
public class ApmJpaCorrelationInterceptor {
private final ElasticApmTracerService tracerService;
public ApmJpaCorrelationInterceptor(ElasticApmTracerService tracerService) {
this.tracerService = tracerService;
}
@EventListener
public void handlePreInsert(PreInsertEvent event) {
trackDatabaseOperation("INSERT", event.getEntity());
}
@EventListener
public void handlePreUpdate(PreUpdateEvent event) {
trackDatabaseOperation("UPDATE", event.getEntity());
}
@EventListener
public void handlePreDelete(PreDeleteEvent event) {
trackDatabaseOperation("DELETE", event.getEntity());
}
@EventListener
public void handlePostLoad(PostLoadEvent event) {
trackDatabaseOperation("SELECT", event.getEntity());
}
private void trackDatabaseOperation(String operation, Object entity) {
co.elastic.apm.api.Span span = tracerService.startSpan("DB " + operation, "db", "postgresql");
if (span != null) {
try {
String entityName = entity.getClass().getSimpleName();
span.setName(operation + " " + entityName);
span.addLabel("db.operation", operation);
span.addLabel("db.entity", entityName);
span.addLabel("db.system", "postgresql");
// Add entity-specific context
addEntityContext(span, entity, operation);
tracerService.setSpanOutcome(true);
} catch (Exception e) {
log.warn("Failed to track database operation", e);
tracerService.setSpanOutcome(false);
} finally {
span.end();
}
}
}
private void addEntityContext(co.elastic.apm.api.Span span, Object entity, String operation) {
if (entity instanceof Identifiable) {
String entityId = ((Identifiable) entity).getId().toString();
span.addLabel("db.entity.id", entityId);
}
if (entity instanceof TrackableEntity) {
TrackableEntity trackable = (TrackableEntity) entity;
span.addLabel("business.domain", trackable.getBusinessDomain());
span.addLabel("tenant.id", trackable.getTenantId());
}
// Add trace correlation
String traceId = tracerService.getCurrentTraceId();
if (traceId != null) {
span.addLabel("correlation.trace_id", traceId);
}
}
}
// Marker interfaces for entity tracking
public interface Identifiable {
Object getId();
}
public interface TrackableEntity {
String getBusinessDomain();
String getTenantId();
}
// Example entity
@Entity
@Table(name = "orders")
@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
public class Order implements Identifiable, TrackableEntity {
@Id
@GeneratedValue(strategy = GenerationType.IDENTITY)
private Long id;
private String orderNumber;
private String customerId;
private BigDecimal totalAmount;
private String status;
private String tenantId;
private LocalDateTime createdAt;
@Override
public String getBusinessDomain() {
return "order-management";
}
}

2. JDBC Template Correlation

@Component
@Slf4j
public class ApmJdbcCorrelationTemplate {
private final JdbcTemplate jdbcTemplate;
private final ElasticApmTracerService tracerService;
public ApmJdbcCorrelationTemplate(JdbcTemplate jdbcTemplate, 
ElasticApmTracerService tracerService) {
this.jdbcTemplate = jdbcTemplate;
this.tracerService = tracerService;
}
/**
* Execute query with APM correlation
*/
public <T> List<T> queryWithCorrelation(String sql, Object[] args, RowMapper<T> rowMapper) {
co.elastic.apm.api.Span span = tracerService.startSpan("DB Query", "db", "postgresql");
try {
span.setName("SQL Query");
span.addLabel("db.statement", sql);
span.addLabel("db.system", "postgresql");
span.addLabel("db.operation", extractDbOperation(sql));
// Add trace correlation
String traceId = tracerService.getCurrentTraceId();
if (traceId != null) {
span.addLabel("correlation.trace_id", traceId);
}
// Execute the query
List<T> result = jdbcTemplate.query(sql, args, rowMapper);
span.addLabel("db.result.count", String.valueOf(result.size()));
tracerService.setSpanOutcome(true);
return result;
} catch (Exception e) {
span.captureException(e);
tracerService.setSpanOutcome(false);
throw new DataAccessException("Database query failed", e) {};
} finally {
span.end();
}
}
/**
* Execute update with APM correlation
*/
public int updateWithCorrelation(String sql, Object... args) {
co.elastic.apm.api.Span span = tracerService.startSpan("DB Update", "db", "postgresql");
try {
span.setName("SQL Update");
span.addLabel("db.statement", sql);
span.addLabel("db.system", "postgresql");
span.addLabel("db.operation", extractDbOperation(sql));
String traceId = tracerService.getCurrentTraceId();
if (traceId != null) {
span.addLabel("correlation.trace_id", traceId);
}
int affectedRows = jdbcTemplate.update(sql, args);
span.addLabel("db.affected.rows", String.valueOf(affectedRows));
tracerService.setSpanOutcome(true);
return affectedRows;
} catch (Exception e) {
span.captureException(e);
tracerService.setSpanOutcome(false);
throw new DataAccessException("Database update failed", e) {};
} finally {
span.end();
}
}
/**
* Execute within a database transaction span
*/
public <T> T executeInDbSpan(String operationName, Supplier<T> operation) {
co.elastic.apm.api.Span span = tracerService.startSpan("DB Transaction", "db", "postgresql");
span.setName(operationName);
try {
T result = operation.get();
tracerService.setSpanOutcome(true);
return result;
} catch (Exception e) {
span.captureException(e);
tracerService.setSpanOutcome(false);
throw e;
} finally {
span.end();
}
}
private String extractDbOperation(String sql) {
if (sql == null) return "unknown";
String normalizedSql = sql.trim().toLowerCase();
if (normalizedSql.startsWith("select")) return "SELECT";
if (normalizedSql.startsWith("insert")) return "INSERT";
if (normalizedSql.startsWith("update")) return "UPDATE";
if (normalizedSql.startsWith("delete")) return "DELETE";
if (normalizedSql.startsWith("create")) return "CREATE";
if (normalizedSql.startsWith("drop")) return "DROP";
if (normalizedSql.startsWith("alter")) return "ALTER";
return "OTHER";
}
}

Message Queue Correlation

1. RabbitMQ Correlation

@Component
@Slf4j
public class ApmMessageCorrelationService {
private final ElasticApmTracerService tracerService;
private final AmqpTemplate amqpTemplate;
private final ObjectMapper objectMapper;
public ApmMessageCorrelationService(ElasticApmTracerService tracerService,
AmqpTemplate amqpTemplate,
ObjectMapper objectMapper) {
this.tracerService = tracerService;
this.amqpTemplate = amqpTemplate;
this.objectMapper = objectMapper;
}
/**
* Send message with APM correlation
*/
public void sendMessageWithCorrelation(String exchange, String routingKey, Object message) {
co.elastic.apm.api.Span span = tracerService.startSpan("Message Publish", "messaging", "rabbitmq");
try {
span.setName("Publish to " + exchange);
span.addLabel("messaging.destination", exchange);
span.addLabel("messaging.destination_kind", "topic");
span.addLabel("messaging.rabbitmq.routing_key", routingKey);
// Create message with trace context
Message amqpMessage = createCorrelatedMessage(message);
// Send message
amqpTemplate.send(exchange, routingKey, amqpMessage);
tracerService.setSpanOutcome(true);
log.debug("Sent correlated message to {}:{}", exchange, routingKey);
} catch (Exception e) {
span.captureException(e);
tracerService.setSpanOutcome(false);
throw new MessageSendException("Failed to send correlated message", e);
} finally {
span.end();
}
}
/**
* Create AMQP message with trace correlation
*/
private Message createCorrelatedMessage(Object payload) throws Exception {
MessageProperties properties = new MessageProperties();
// Inject trace context into message headers
Map<String, String> traceHeaders = new HashMap<>();
tracerService.injectTraceContext(traceHeaders);
traceHeaders.forEach((key, value) -> properties.setHeader(key, value));
// Add correlation metadata
properties.setHeader("x-correlation-trace-id", tracerService.getCurrentTraceId());
properties.setHeader("x-correlation-transaction-id", tracerService.getCurrentTransactionId());
properties.setHeader("x-correlation-timestamp", Instant.now().toString());
properties.setHeader("x-correlation-service", "order-service");
byte[] body = objectMapper.writeValueAsBytes(payload);
return new Message(body, properties);
}
/**
* Process message with correlation context
*/
public <T> void processMessageWithCorrelation(Message message, Class<T> payloadType, 
Function<T, Boolean> processor) {
// Extract trace context from message
co.elastic.apm.api.Transaction transaction = extractTransactionFromMessage(message);
if (transaction == null) {
transaction = tracerService.startTransaction("Message Process", "messaging", null);
}
try {
configureMessageTransaction(transaction, message);
// Extract payload
T payload = extractPayload(message, payloadType);
// Process the message
boolean success = processor.apply(payload);
if (success) {
tracerService.setTransactionOutcome(true);
log.debug("Successfully processed correlated message");
} else {
tracerService.setTransactionOutcome(false);
}
} catch (Exception e) {
transaction.captureException(e);
tracerService.setTransactionOutcome(false);
throw new MessageProcessingException("Failed to process correlated message", e);
} finally {
transaction.end();
}
}
private co.elastic.apm.api.Transaction extractTransactionFromMessage(Message message) {
MessageProperties properties = message.getMessageProperties();
Map<String, String> headers = new HashMap<>();
if (properties.getHeaders() != null) {
properties.getHeaders().forEach((key, value) -> {
if (value instanceof String) {
headers.put(key, (String) value);
}
});
}
return tracerService.extractTraceContext(headers);
}
private void configureMessageTransaction(co.elastic.apm.api.Transaction transaction, Message message) {
MessageProperties properties = message.getMessageProperties();
transaction.setName("Process from " + properties.getConsumerQueue());
transaction.addLabel("messaging.system", "rabbitmq");
transaction.addLabel("messaging.destination", properties.getReceivedExchange());
transaction.addLabel("messaging.destination_kind", "topic");
transaction.addLabel("messaging.rabbitmq.routing_key", properties.getReceivedRoutingKey());
transaction.addLabel("messaging.rabbitmq.queue", properties.getConsumerQueue());
// Add correlation context
String correlationId = (String) properties.getHeader("x-correlation-id");
if (correlationId != null) {
transaction.addLabel("correlation.id", correlationId);
}
}
private <T> T extractPayload(Message message, Class<T> payloadType) {
try {
return objectMapper.readValue(message.getBody(), payloadType);
} catch (Exception e) {
throw new MessageProcessingException("Failed to extract message payload", e);
}
}
}

Business Transaction Correlation

1. Order Processing with Correlation

@Service
@Slf4j
public class CorrelatedOrderService {
private final ElasticApmTracerService tracerService;
private final ApmJdbcCorrelationTemplate jdbcTemplate;
private final ApmCorrelatedHttpClient httpClient;
private final ApmMessageCorrelationService messageService;
public CorrelatedOrderService(ElasticApmTracerService tracerService,
ApmJdbcCorrelationTemplate jdbcTemplate,
ApmCorrelatedHttpClient httpClient,
ApmMessageCorrelationService messageService) {
this.tracerService = tracerService;
this.jdbcTemplate = jdbcTemplate;
this.httpClient = httpClient;
this.messageService = messageService;
}
/**
* Process order with full APM correlation
*/
@Transactional
public Order processOrder(CreateOrderRequest request) {
co.elastic.apm.api.Transaction transaction = tracerService.getCurrentTransaction();
try {
// Add business context to transaction
addBusinessContext(transaction, request);
// Step 1: Validate order
validateOrder(request);
// Step 2: Check inventory
checkInventory(request);
// Step 3: Process payment
processPayment(request);
// Step 4: Create order record
Order order = createOrderRecord(request);
// Step 5: Send notifications
sendOrderNotifications(order);
tracerService.setTransactionOutcome(true);
log.info("Successfully processed order: {}", order.getOrderNumber());
return order;
} catch (Exception e) {
transaction.captureException(e);
tracerService.setTransactionOutcome(false);
log.error("Order processing failed", e);
throw new OrderProcessingException("Order processing failed", e);
}
}
private void validateOrder(CreateOrderRequest request) {
co.elastic.apm.api.Span span = tracerService.startSpan("Validate Order", "business", "validation");
try {
span.addLabel("order.customer_id", request.getCustomerId());
span.addLabel("order.item_count", String.valueOf(request.getItems().size()));
// Validation logic
if (request.getItems().isEmpty()) {
throw new ValidationException("Order must contain at least one item");
}
if (request.getTotalAmount().compareTo(BigDecimal.ZERO) <= 0) {
throw new ValidationException("Order total must be positive");
}
tracerService.setSpanOutcome(true);
} catch (Exception e) {
span.captureException(e);
tracerService.setSpanOutcome(false);
throw e;
} finally {
span.end();
}
}
private void checkInventory(CreateOrderRequest request) {
co.elastic.apm.api.Span span = tracerService.startSpan("Check Inventory", "business", "inventory");
try {
span.addLabel("inventory.check.items", String.valueOf(request.getItems().size()));
// Call inventory service with correlation
InventoryCheckRequest inventoryRequest = createInventoryRequest(request);
InventoryCheckResponse response = httpClient.callServiceWithCorrelation(
"inventory-service", "/api/inventory/check", inventoryRequest, InventoryCheckResponse.class)
.block();
if (!response.isAvailable()) {
throw new InventoryException("Insufficient inventory for order");
}
span.addLabel("inventory.available", "true");
tracerService.setSpanOutcome(true);
} catch (Exception e) {
span.captureException(e);
tracerService.setSpanOutcome(false);
throw e;
} finally {
span.end();
}
}
private void processPayment(CreateOrderRequest request) {
co.elastic.apm.api.Span span = tracerService.startSpan("Process Payment", "business", "payment");
try {
span.addLabel("payment.amount", request.getTotalAmount().toString());
span.addLabel("payment.currency", request.getCurrency());
// Call payment service with correlation
PaymentRequest paymentRequest = createPaymentRequest(request);
PaymentResponse response = httpClient.callServiceWithCorrelation(
"payment-service", "/api/payments/process", paymentRequest, PaymentResponse.class)
.block();
if (!response.isSuccess()) {
throw new PaymentException("Payment processing failed: " + response.getErrorMessage());
}
span.addLabel("payment.transaction_id", response.getTransactionId());
tracerService.setSpanOutcome(true);
} catch (Exception e) {
span.captureException(e);
tracerService.setSpanOutcome(false);
throw e;
} finally {
span.end();
}
}
private Order createOrderRecord(CreateOrderRequest request) {
return jdbcTemplate.executeInDbSpan("Create Order Record", () -> {
String orderNumber = generateOrderNumber();
Order order = Order.builder()
.orderNumber(orderNumber)
.customerId(request.getCustomerId())
.totalAmount(request.getTotalAmount())
.status("CONFIRMED")
.tenantId(request.getTenantId())
.createdAt(LocalDateTime.now())
.build();
// Save order
String sql = "INSERT INTO orders (order_number, customer_id, total_amount, status, tenant_id, created_at) VALUES (?, ?, ?, ?, ?, ?)";
jdbcTemplate.updateWithCorrelation(sql, 
order.getOrderNumber(),
order.getCustomerId(),
order.getTotalAmount(),
order.getStatus(),
order.getTenantId(),
order.getCreatedAt()
);
tracerService.addLabel("order.number", orderNumber);
return order;
});
}
private void sendOrderNotifications(Order order) {
co.elastic.apm.api.Span span = tracerService.startSpan("Send Notifications", "business", "messaging");
try {
// Send order confirmation message
OrderConfirmedEvent event = OrderConfirmedEvent.builder()
.orderId(order.getId())
.orderNumber(order.getOrderNumber())
.customerId(order.getCustomerId())
.totalAmount(order.getTotalAmount())
.confirmedAt(LocalDateTime.now())
.build();
messageService.sendMessageWithCorrelation(
"order-events", 
"order.confirmed", 
event
);
span.addLabel("notification.sent", "true");
tracerService.setSpanOutcome(true);
} catch (Exception e) {
span.captureException(e);
tracerService.setSpanOutcome(false);
log.warn("Failed to send order notifications", e);
// Don't throw - notifications are non-critical
} finally {
span.end();
}
}
private void addBusinessContext(co.elastic.apm.api.Transaction transaction, CreateOrderRequest request) {
transaction.addLabel("business.process", "order-processing");
transaction.addLabel("business.customer_id", request.getCustomerId());
transaction.addLabel("business.tenant_id", request.getTenantId());
transaction.addLabel("business.order_amount", request.getTotalAmount().toString());
transaction.addLabel("business.currency", request.getCurrency());
}
private String generateOrderNumber() {
return "ORD-" + System.currentTimeMillis() + "-" + 
ThreadLocalRandom.current().nextInt(1000, 9999);
}
private InventoryCheckRequest createInventoryRequest(CreateOrderRequest orderRequest) {
return InventoryCheckRequest.builder()
.items(orderRequest.getItems())
.tenantId(orderRequest.getTenantId())
.build();
}
private PaymentRequest createPaymentRequest(CreateOrderRequest orderRequest) {
return PaymentRequest.builder()
.amount(orderRequest.getTotalAmount())
.currency(orderRequest.getCurrency())
.customerId(orderRequest.getCustomerId())
.paymentMethod(orderRequest.getPaymentMethod())
.build();
}
}

Custom Correlation Service

1. Correlation ID Service

@Component
@Slf4j
public class CorrelationIdService {
private static final String CORRELATION_ID_HEADER = "X-Correlation-ID";
private static final String CORRELATION_ID_LABEL = "correlation.id";
private final ElasticApmTracerService tracerService;
public CorrelationIdService(ElasticApmTracerService tracerService) {
this.tracerService = tracerService;
}
/**
* Generate new correlation ID
*/
public String generateCorrelationId() {
return "corr-" + UUID.randomUUID().toString();
}
/**
* Get current correlation ID from transaction
*/
public String getCurrentCorrelationId() {
// In real implementation, you might store this in a ThreadLocal
// or extract from the current transaction context
co.elastic.apm.api.Transaction transaction = tracerService.getCurrentTransaction();
if (transaction != null) {
// This would require custom implementation to get labels
// For demo purposes, we'll generate a new one
return generateCorrelationId();
}
return generateCorrelationId();
}
/**
* Set correlation ID in current transaction
*/
public void setCorrelationId(String correlationId) {
tracerService.addLabel(CORRELATION_ID_LABEL, correlationId);
}
/**
* Inject correlation ID into headers
*/
public void injectCorrelationId(Map<String, String> headers) {
String correlationId = getCurrentCorrelationId();
headers.put(CORRELATION_ID_HEADER, correlationId);
// Also ensure it's set in the transaction
setCorrelationId(correlationId);
}
/**
* Extract correlation ID from headers
*/
public String extractCorrelationId(Map<String, String> headers) {
String correlationId = headers.get(CORRELATION_ID_HEADER);
if (correlationId == null || correlationId.isEmpty()) {
correlationId = generateCorrelationId();
}
setCorrelationId(correlationId);
return correlationId;
}
/**
* Create correlated business transaction
*/
public CorrelatedTransaction startCorrelatedTransaction(String name, String type, Map<String, String> context) {
co.elastic.apm.api.Transaction transaction = tracerService.startTransaction(name, type, null);
String correlationId = getCurrentCorrelationId();
// Add context labels
if (context != null) {
context.forEach(tracerService::addLabel);
}
tracerService.addLabel(CORRELATION_ID_LABEL, correlationId);
return CorrelatedTransaction.builder()
.transaction(transaction)
.correlationId(correlationId)
.traceId(tracerService.getCurrentTraceId())
.build();
}
@Data
@Builder
public static class CorrelatedTransaction {
private co.elastic.apm.api.Transaction transaction;
private String correlationId;
private String traceId;
public void end(boolean success) {
if (transaction != null) {
transaction.setOutcome(success ? "success" : "failure");
transaction.end();
}
}
}
}

2. Business Context Propagator

@Component
@Slf4j
public class BusinessContextPropagator {
private final ElasticApmTracerService tracerService;
private final CorrelationIdService correlationIdService;
public BusinessContextPropagator(ElasticApmTracerService tracerService,
CorrelationIdService correlationIdService) {
this.tracerService = tracerService;
this.correlationIdService = correlationIdService;
}
/**
* Add user context to current transaction
*/
public void addUserContext(String userId, String sessionId, String userRole) {
tracerService.addLabel("user.id", userId);
tracerService.addLabel("user.session_id", sessionId);
tracerService.addLabel("user.role", userRole);
log.debug("Added user context: userId={}, role={}", userId, userRole);
}
/**
* Add tenant context to current transaction
*/
public void addTenantContext(String tenantId, String tenantName, String environment) {
tracerService.addLabel("tenant.id", tenantId);
tracerService.addLabel("tenant.name", tenantName);
tracerService.addLabel("tenant.environment", environment);
log.debug("Added tenant context: tenantId={}, environment={}", tenantId, environment);
}
/**
* Add business process context
*/
public void addBusinessProcessContext(String processName, String processId, String step) {
tracerService.addLabel("business.process", processName);
tracerService.addLabel("business.process_id", processId);
tracerService.addLabel("business.process_step", step);
log.debug("Added business process context: process={}, step={}", processName, step);
}
/**
* Add financial context
*/
public void addFinancialContext(String currency, BigDecimal amount, String transactionType) {
tracerService.addLabel("financial.currency", currency);
tracerService.addLabel("financial.amount", amount.toString());
tracerService.addLabel("financial.transaction_type", transactionType);
}
/**
* Propagate business context to outgoing requests
*/
public void propagateBusinessContext(Map<String, String> headers) {
// Inject correlation ID
correlationIdService.injectCorrelationId(headers);
// Add business context headers
co.elastic.apm.api.Transaction transaction = tracerService.getCurrentTransaction();
if (transaction != null) {
// In real implementation, you would extract labels from transaction
// and add them as headers
headers.put("X-Business-Context", "active");
}
log.debug("Propagated business context to headers");
}
/**
* Extract business context from incoming requests
*/
public void extractBusinessContext(Map<String, String> headers) {
// Extract correlation ID
String correlationId = correlationIdService.extractCorrelationId(headers);
// Extract other business context from headers
String userId = headers.get("X-User-ID");
String tenantId = headers.get("X-Tenant-ID");
String userRole = headers.get("X-User-Role");
if (userId != null) {
addUserContext(userId, headers.get("X-Session-ID"), userRole);
}
if (tenantId != null) {
addTenantContext(tenantId, headers.get("X-Tenant-Name"), headers.get("X-Environment"));
}
log.debug("Extracted business context: correlationId={}, userId={}", correlationId, userId);
}
/**
* Create business transaction with full context
*/
public CorrelationIdService.CorrelatedTransaction startBusinessTransaction(
String name, BusinessContext context) {
Map<String, String> labels = new HashMap<>();
labels.put("business.domain", context.getBusinessDomain());
labels.put("business.operation", context.getOperation());
labels.put("business.criticality", context.getCriticality());
if (context.getUserId() != null) {
labels.put("user.id", context.getUserId());
}
if (context.getTenantId() != null) {
labels.put("tenant.id", context.getTenantId());
}
return correlationIdService.startCorrelatedTransaction(name, "business", labels);
}
@Data
@Builder
public static class BusinessContext {
private String businessDomain;
private String operation;
private String criticality; // LOW, MEDIUM, HIGH, CRITICAL
private String userId;
private String tenantId;
private String sessionId;
private String userRole;
}
}

REST Controller with APM Correlation

@RestController
@RequestMapping("/api/orders")
@Slf4j
public class CorrelatedOrderController {
private final CorrelatedOrderService orderService;
private final ElasticApmTracerService tracerService;
private final BusinessContextPropagator contextPropagator;
private final CorrelationIdService correlationIdService;
public CorrelatedOrderController(CorrelatedOrderService orderService,
ElasticApmTracerService tracerService,
BusinessContextPropagator contextPropagator,
CorrelationIdService correlationIdService) {
this.orderService = orderService;
this.tracerService = tracerService;
this.contextPropagator = contextPropagator;
this.correlationIdService = correlationIdService;
}
@PostMapping
public ResponseEntity<OrderResponse> createOrder(@RequestBody CreateOrderRequest request,
HttpServletRequest httpRequest) {
// Extract business context from headers
Map<String, String> headers = extractHeaders(httpRequest);
contextPropagator.extractBusinessContext(headers);
CorrelationIdService.CorrelatedTransaction transaction = 
contextPropagator.startBusinessTransaction(
"Create Order",
BusinessContextPropagator.BusinessContext.builder()
.businessDomain("order-management")
.operation("create-order")
.criticality("HIGH")
.userId(request.getCustomerId())
.tenantId(request.getTenantId())
.build()
);
try {
// Process the order
Order order = orderService.processOrder(request);
OrderResponse response = OrderResponse.builder()
.orderId(order.getId())
.orderNumber(order.getOrderNumber())
.status(order.getStatus())
.correlationId(transaction.getCorrelationId())
.traceId(transaction.getTraceId())
.message("Order created successfully")
.build();
transaction.end(true);
return ResponseEntity.ok(response);
} catch (Exception e) {
transaction.end(false);
throw e;
}
}
@GetMapping("/{orderId}")
public ResponseEntity<OrderResponse> getOrder(@PathVariable Long orderId,
HttpServletRequest httpRequest) {
// Extract business context
Map<String, String> headers = extractHeaders(httpRequest);
contextPropagator.extractBusinessContext(headers);
CorrelationIdService.CorrelatedTransaction transaction = 
contextPropagator.startBusinessTransaction(
"Get Order",
BusinessContextPropagator.BusinessContext.builder()
.businessDomain("order-management")
.operation("get-order")
.criticality("LOW")
.build()
);
try {
// In real implementation, fetch order from service
Order order = findOrderById(orderId);
if (order == null) {
transaction.end(false);
return ResponseEntity.notFound().build();
}
OrderResponse response = OrderResponse.builder()
.orderId(order.getId())
.orderNumber(order.getOrderNumber())
.status(order.getStatus())
.correlationId(transaction.getCorrelationId())
.traceId(transaction.getTraceId())
.build();
transaction.end(true);
return ResponseEntity.ok(response);
} catch (Exception e) {
transaction.end(false);
throw e;
}
}
@GetMapping("/correlation/current")
public ResponseEntity<CorrelationInfo> getCurrentCorrelation() {
CorrelationInfo info = CorrelationInfo.builder()
.correlationId(correlationIdService.getCurrentCorrelationId())
.traceId(tracerService.getCurrentTraceId())
.transactionId(tracerService.getCurrentTransactionId())
.serviceName("order-service")
.timestamp(Instant.now())
.build();
return ResponseEntity.ok(info);
}
private Map<String, String> extractHeaders(HttpServletRequest request) {
Map<String, String> headers = new HashMap<>();
Enumeration<String> headerNames = request.getHeaderNames();
while (headerNames.hasMoreElements()) {
String headerName = headerNames.nextElement();
headers.put(headerName, request.getHeader(headerName));
}
return headers;
}
private Order findOrderById(Long orderId) {
// Mock implementation
return Order.builder()
.id(orderId)
.orderNumber("ORD-12345")
.status("CONFIRMED")
.build();
}
@Data
@Builder
public static class OrderResponse {
private Long orderId;
private String orderNumber;
private String status;
private String correlationId;
private String traceId;
private String message;
}
@Data
@Builder
public static class CorrelationInfo {
private String correlationId;
private String traceId;
private String transactionId;
private String serviceName;
private Instant timestamp;
}
}

Testing APM Correlation

1. Correlation Test Suite

@SpringBootTest
@ActiveProfiles("test")
class ElasticApmCorrelationTest {
@Autowired
private ElasticApmTracerService tracerService;
@Autowired
private CorrelationIdService correlationIdService;
@Autowired
private BusinessContextPropagator contextPropagator;
@MockBean
private ApmCorrelatedHttpClient httpClient;
@Test
void testCorrelationIdGeneration() {
// Given
String correlationId = correlationIdService.generateCorrelationId();
// When
correlationIdService.setCorrelationId(correlationId);
String retrievedId = correlationIdService.getCurrentCorrelationId();
// Then
assertNotNull(correlationId);
assertTrue(correlationId.startsWith("corr-"));
assertEquals(correlationId, retrievedId);
}
@Test
void testTraceContextInjection() {
// Given
Map<String, String> headers = new HashMap<>();
// When
tracerService.injectTraceContext(headers);
// Then
assertNotNull(headers.get("traceparent"));
assertNotNull(headers.get("tracestate"));
}
@Test
void testBusinessContextPropagation() {
// Given
Map<String, String> headers = new HashMap<>();
// When
contextPropagator.addUserContext("user-123", "session-456", "ADMIN");
contextPropagator.addTenantContext("tenant-789", "Acme Corp", "production");
contextPropagator.propagateBusinessContext(headers);
// Then
assertNotNull(headers.get("X-Correlation-ID"));
assertEquals("active", headers.get("X-Business-Context"));
}
@Test
void testCorrelatedTransaction() {
// Given
BusinessContextPropagator.BusinessContext context = 
BusinessContextPropagator.BusinessContext.builder()
.businessDomain("order-management")
.operation("test-operation")
.criticality("HIGH")
.userId("test-user")
.build();
// When
CorrelationIdService.CorrelatedTransaction transaction = 
contextPropagator.startBusinessTransaction("Test Transaction", context);
// Then
assertNotNull(transaction);
assertNotNull(transaction.getCorrelationId());
assertNotNull(transaction.getTraceId());
// Cleanup
transaction.end(true);
}
}

Monitoring and Metrics

1. APM Metrics Collector

@Component
@Slf4j
public class ApmMetricsCollector {
private final MeterRegistry meterRegistry;
private final ElasticApmTracerService tracerService;
public ApmMetricsCollector(MeterRegistry meterRegistry, 
ElasticApmTracerService tracerService) {
this.meterRegistry = meterRegistry;
this.tracerService = tracerService;
}
/**
* Record business transaction metrics
*/
public void recordBusinessTransaction(String transactionName, boolean success, Duration duration) {
Counter.builder("business.transactions")
.tag("name", transactionName)
.tag("outcome", success ? "success" : "failure")
.register(meterRegistry)
.increment();
Timer.builder("business.transaction.duration")
.tag("name", transactionName)
.register(meterRegistry)
.record(duration);
log.debug("Recorded business transaction: {}, success: {}, duration: {}ms", 
transactionName, success, duration.toMillis());
}
/**
* Record external service call metrics
*/
public void recordExternalServiceCall(String serviceName, String endpoint, 
boolean success, Duration duration) {
Counter.builder("external.service.calls")
.tag("service", serviceName)
.tag("endpoint", endpoint)
.tag("success", String.valueOf(success))
.register(meterRegistry)
.increment();
Timer.builder("external.service.duration")
.tag("service", serviceName)
.tag("endpoint", endpoint)
.register(meterRegistry)
.record(duration);
}
/**
* Record database operation metrics
*/
public void recordDatabaseOperation(String operation, String table, 
boolean success, Duration duration) {
Counter.builder("database.operations")
.tag("operation", operation)
.tag("table", table)
.tag("success", String.valueOf(success))
.register(meterRegistry)
.increment();
Timer.builder("database.operation.duration")
.tag("operation", operation)
.tag("table", table)
.register(meterRegistry)
.record(duration);
}
/**
* Record correlation metrics
*/
public void recordCorrelationEvent(String eventType, String source, String target) {
Counter.builder("correlation.events")
.tag("event_type", eventType)
.tag("source", source)
.tag("target", target)
.register(meterRegistry)
.increment();
log.debug("Recorded correlation event: {} from {} to {}", eventType, source, target);
}
}

Conclusion

Elastic APM correlation provides powerful observability capabilities for Java applications. By properly correlating traces across services, databases, and message queues, you gain comprehensive visibility into your application's behavior and performance.

Key Implementation Patterns:

  • HTTP Correlation: Automatic trace context propagation for REST APIs
  • Database Correlation: Span creation for SQL operations and JPA events
  • Message Queue Correlation: Trace context propagation through messaging systems
  • Business Transaction Correlation: Custom spans for business processes
  • Context Propagation: Business context and correlation ID management

Best Practices:

  • Consistent Naming: Use meaningful names for transactions and spans
  • Context Enrichment: Add business context to traces for better debugging
  • Performance Awareness: Be mindful of span creation overhead
  • Error Handling: Properly capture and correlate exceptions
  • Testing: Comprehensive testing of correlation across service boundaries

Use Cases:

  • Microservices Architecture: End-to-end request tracing
  • Complex Business Processes: Correlation of multi-step transactions
  • Performance Monitoring: Identifying bottlenecks across services
  • Error Analysis: Correlating errors with specific traces and contexts
  • Capacity Planning: Understanding service dependencies and load patterns

By implementing comprehensive Elastic APM correlation, you can achieve full-stack observability, enabling effective monitoring, debugging, and optimization of your distributed Java applications.

Category : JAVA

Leave a Reply

Your email address will not be published. Required fields are marked *


Macro Nepal Helper