Kibana version:
v 7.12.1
Elasticsearch version:
7.12.1
APM Server version:
7.12.1
APM Agent language and version:
"co.elastic.apm:apm-agent-attach:1.25.0"
Original install method (e.g. download page, yum, deb, from source, etc.) and version:
@Setter
@Configuration
@ConfigurationProperties(prefix = "app.elastic-apm")
public class ElasticApmConfiguration {
private static final String SERVER_URL_KEY = "server_url";
private String serverUrl;
private static final String SERVICE_NAME_KEY = "service_name";
private String serviceName;
private static final String ENVIRONMENT_KEY = "environment";
private String environment;
private static final String APPLICATION_PACKAGES_KEY = "application_packages";
private String applicationPackages;
@PostConstruct
public void init() {
Map<String, String> apmProps = new HashMap<>();
apmProps.put(SERVER_URL_KEY, serverUrl);
apmProps.put(SERVICE_NAME_KEY, serviceName);
apmProps.put(ENVIRONMENT_KEY, environment);
apmProps.put(APPLICATION_PACKAGES_KEY, applicationPackages);
ElasticApmAttacher.attach(apmProps);
}
}
Description of the problem including expected versus actual behavior. Please include screenshots (if relevant):
Hi, I have a simple spring boot + kafka + apm app. I have a @KafkaListener like this:
@Slf4j
@EnableKafka
@Component
@RequiredArgsConstructor
public class CampaignCleanedListener {
...deps...
@KafkaListener(topics = "${app.kafka.topics.mytopic}")
public void consume(@Payload String input, Acknowledgment acknowledgment) {
....my logic....
acknowledgment.acknowledge();
}
}
Everything works smoothly I see in my kibana in the APM graphs that average time to process a message is about 500 ms.
However when I add my custom ConcurrentKafkaListenerContainerFactory like that:
@Configuration
public class MyConfiguration {
@Bean
public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
KafkaAutoConfiguration kafkaAutoConfiguration,
ObjectProvider<DefaultKafkaConsumerFactoryCustomizer> customizers) {
ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
configurer.configure(factory, (ConsumerFactory<Object, Object>) kafkaAutoConfiguration.kafkaConsumerFactory(customizers));
factory.setBatchListener(true);
factory.setBatchErrorHandler(new RetryingBatchErrorHandler(new ExponentialBackOff(), null));
return factory;
}
}
suddenly it stops working properly and shows some false values like 3 μs for a message. I am not even sure what it measures. I would like to ask whether it some bug or it works as expected and if so how to make the apm working with custom ConcurrentKafkaListenerContainerFactory
Steps to reproduce:
- Create spring boot app with default @KafkaListener
- Add custom bean for
ConcurrentKafkaListenerContainerFactory - Observe that in the apm graph average times are 100 times less than they should be