org.springframework.context.annotation.ConfigurationClassParser.getConfigurationClasses()讀所有@Configuration類，傳入org.springframework.context.annotation.ConfigurationClassBeanDefinitionReader.loadBeanDefinitions()進行@Configuration對象定義加載，加載@Configuration類定義@Bean方法、@Import等資源，如使用@EnableKafka加載KafkaBootstrapConfiguration然后加載KafkaListenerAnnotationBeanPostProcessor如下列stack所示：
Thread [main] (Suspended (breakpoint at line 188 in ConfigurationClass))
owns: Object (id=120)
ConfigurationClass.addBeanMethod(BeanMethod) line: 188
ConfigurationClassParser.doProcessConfigurationClass(ConfigurationClass, ConfigurationClassParser$SourceClass) line: 316
ConfigurationClassParser.processConfigurationClass(ConfigurationClass) line: 244
ConfigurationClassParser.processImports(ConfigurationClass, SourceClass, Collection<SourceClass>, boolean) line: 606
ConfigurationClassParser.doProcessConfigurationClass(ConfigurationClass, ConfigurationClassParser$SourceClass) line: 299
ConfigurationClassParser.processConfigurationClass(ConfigurationClass) line: 244
ConfigurationClassParser.parse(String, String) line: 189
ConfigurationClassParser.doProcessConfigurationClass(ConfigurationClass, ConfigurationClassParser$SourceClass) line: 292
ConfigurationClassParser.processConfigurationClass(ConfigurationClass) line: 244
ConfigurationClassParser.parse(AnnotationMetadata, String) line: 197
ConfigurationClassParser.parse(Set<BeanDefinitionHolder>) line: 165
ConfigurationClassPostProcessor.processConfigBeanDefinitions(BeanDefinitionRegistry) line: 308
ConfigurationClassPostProcessor.postProcessBeanDefinitionRegistry(BeanDefinitionRegistry) line: 229
PostProcessorRegistrationDelegate.invokeBeanDefinitionRegistryPostProcessors(Collection<BeanDefinitionRegistryPostProcessor>, BeanDefinitionRegistry) line: 271
PostProcessorRegistrationDelegate.invokeBeanFactoryPostProcessors(ConfigurableListableBeanFactory, List<BeanFactoryPostProcessor>) line: 94
AnnotationConfigApplicationContext(AbstractApplicationContext).invokeBeanFactoryPostProcessors(ConfigurableListableBeanFactory) line: 686
AnnotationConfigApplicationContext(AbstractApplicationContext).refresh() line: 524
SpringApplication.refresh(ApplicationContext) line: 750
SpringApplication.refreshContext(ConfigurableApplicationContext) line: 386
SpringApplication.run(String...) line: 327
SpringApplication.run(Class<?>[], String[]) line: 1245
SpringApplication.run(Class<?>, String...) line: 1233
Application.main(String[]) line: 51
KafkaListenerAnnotationBeanPostProcessor對象定義注冊在DefaultListableBeanFactory.registerBeanDefinition()，使用名稱為KafkaListenerConfigUtils.KAFKA_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME，如下列stack:
Thread [main] (Suspended (breakpoint at line 779 in DefaultListableBeanFactory))
owns: Object (id=130)
DefaultListableBeanFactory.registerBeanDefinition(String, BeanDefinition) line: 779
ConfigurationClassBeanDefinitionReader.loadBeanDefinitionsForBeanMethod(BeanMethod) line: 262
ConfigurationClassBeanDefinitionReader.loadBeanDefinitionsForConfigurationClass(ConfigurationClass, ConfigurationClassBeanDefinitionReader$TrackedConditionEvaluator) line: 141
ConfigurationClassBeanDefinitionReader.loadBeanDefinitions(Set<ConfigurationClass>) line: 117
ConfigurationClassPostProcessor.processConfigBeanDefinitions(BeanDefinitionRegistry) line: 320
ConfigurationClassPostProcessor.postProcessBeanDefinitionRegistry(BeanDefinitionRegistry) line: 229
PostProcessorRegistrationDelegate.invokeBeanDefinitionRegistryPostProcessors(Collection<BeanDefinitionRegistryPostProcessor>, BeanDefinitionRegistry) line: 271
PostProcessorRegistrationDelegate.invokeBeanFactoryPostProcessors(ConfigurableListableBeanFactory, List<BeanFactoryPostProcessor>) line: 94
AnnotationConfigApplicationContext(AbstractApplicationContext).invokeBeanFactoryPostProcessors(ConfigurableListableBeanFactory) line: 686
AnnotationConfigApplicationContext(AbstractApplicationContext).refresh() line: 524
SpringApplication.refresh(ApplicationContext) line: 750
SpringApplication.refreshContext(ConfigurableApplicationContext) line: 386
SpringApplication.run(String...) line: 327
SpringApplication.run(Class<?>[], String[]) line: 1245
SpringApplication.run(Class<?>, String...) line: 1233
Application.main(String[]) line: 51
KafkaListenerAnnotationBeanPostProcessor對象定義注冊后，在PostProcessorRegistrationDelegate.registerBeanPostProcessors()實例化，將KafkaListenerAnnotationBeanPostProcessor對象添加到DefaultListableBeanFactory.beanPostProcessors保存，用於后續注釋了@KafkaListener的對象實例化后處理事件的調用。KafkaListenerAnnotationBeanPostProcessor在AnnotationConfigApplicationContext(AbstractApplicationContext).registerBeanPostProcessors(beanFactory)注冊，如下圖stack:
Thread [main] (Suspended (breakpoint at line 848 in AbstractBeanFactory))
owns: Object (id=361)
DefaultListableBeanFactory(AbstractBeanFactory).addBeanPostProcessor(BeanPostProcessor) line: 848
PostProcessorRegistrationDelegate.registerBeanPostProcessors(ConfigurableListableBeanFactory, List<BeanPostProcessor>) line: 293
PostProcessorRegistrationDelegate.registerBeanPostProcessors(ConfigurableListableBeanFactory, AbstractApplicationContext) line: 231
AnnotationConfigApplicationContext(AbstractApplicationContext).registerBeanPostProcessors(ConfigurableListableBeanFactory) line: 702
AnnotationConfigApplicationContext(AbstractApplicationContext).refresh() line: 527
SpringApplication.refresh(ApplicationContext) line: 750
SpringApplication.refreshContext(ConfigurableApplicationContext) line: 386
SpringApplication.run(String...) line: 327
SpringApplication.run(Class<?>[], String[]) line: 1245
SpringApplication.run(Class<?>, String...) line: 1233
Application.main(String[]) line: 51
AnnotationConfigApplicationContext.finishBeanFactoryInitialization()用戶級對象實例化后處理事件的調用，如觸發KafkaListenerAnnotationBeanPostProcessor<K,V>.processListener調用，邏輯是注釋@KafkaListener方法在ConcurrentMessageListenerContainer注冊成監聽事件，有接收到kafka消息時補調用。@KafkaListener方法封閉一個MethodKafkaListenerEndpoint對象並注冊執行
KafkaListenerAnnotationBeanPostProcessor.registerEndpoint(endpoint, factory)，注冊邏輯MethodKafkaListenerEndpoint.
createMessageListener()生成MessagingMessageListenerAdapter對象，factory對象是KafkaListenerContainerFactory可以生成MessageListenerContainer對象，MessagingMessageListenerAdapter對象通過MessageListenerContainer.setupMessageListener()注冊。Container監聽線程發現有消息時執行KafkaMessageListenerContainer$ListenerConsumer.doInvokeRecordListener()，執行MessagingMessageListenerAdapter封裝@KafkaListener注釋方法。UML類圖https://www.lucidchart.com/documents/view/d5e33d39-b5a7-4857-a2c6-60b4c7fd0139
Thread [main] (Suspended (breakpoint at line 64 in MethodKafkaListenerEndpoint))
owns: ConcurrentHashMap<K,V> (id=79)
owns: Object (id=80)
MethodKafkaListenerEndpoint<K,V>.setBean(Object) line: 64
KafkaListenerAnnotationBeanPostProcessor<K,V>.processListener(MethodKafkaListenerEndpoint<?,?>, KafkaListener, Object, Object, String) line: 377
KafkaListenerAnnotationBeanPostProcessor<K,V>.processKafkaListener(KafkaListener, Method, Object, String) line: 340
KafkaListenerAnnotationBeanPostProcessor<K,V>.postProcessAfterInitialization(Object, String) line: 270
DefaultListableBeanFactory(AbstractAutowireCapableBeanFactory).applyBeanPostProcessorsAfterInitialization(Object, String) line: 435
DefaultListableBeanFactory(AbstractAutowireCapableBeanFactory).initializeBean(String, Object, RootBeanDefinition) line: 1721
DefaultListableBeanFactory(AbstractAutowireCapableBeanFactory).doCreateBean(String, RootBeanDefinition, Object[]) line: 581
DefaultListableBeanFactory(AbstractAutowireCapableBeanFactory).createBean(String, RootBeanDefinition, Object[]) line: 498
DefaultListableBeanFactory(AbstractBeanFactory).lambda$doGetBean$0(String, RootBeanDefinition, Object[]) line: 305
2099051403.getObject() line: not available
DefaultListableBeanFactory(DefaultSingletonBeanRegistry).getSingleton(String, ObjectFactory<?>) line: 233
DefaultListableBeanFactory(AbstractBeanFactory).doGetBean(String, Class<T>, Object[], boolean) line: 303
DefaultListableBeanFactory(AbstractBeanFactory).getBean(String, Class<T>) line: 198
CommonAnnotationBeanPostProcessor.autowireResource(BeanFactory, CommonAnnotationBeanPostProcessor$LookupElement, String) line: 512
CommonAnnotationBeanPostProcessor.getResource(CommonAnnotationBeanPostProcessor$LookupElement, String) line: 483
CommonAnnotationBeanPostProcessor$ResourceElement.getResourceToInject(Object, String) line: 617
CommonAnnotationBeanPostProcessor$ResourceElement(InjectionMetadata$InjectedElement).inject(Object, String, PropertyValues) line: 172
InjectionMetadata.inject(Object, String, PropertyValues) line: 89
CommonAnnotationBeanPostProcessor.postProcessPropertyValues(PropertyValues, PropertyDescriptor[], Object, String) line: 317
DefaultListableBeanFactory(AbstractAutowireCapableBeanFactory).populateBean(String, RootBeanDefinition, BeanWrapper) line: 1353
DefaultListableBeanFactory(AbstractAutowireCapableBeanFactory).doCreateBean(String, RootBeanDefinition, Object[]) line: 579
DefaultListableBeanFactory(AbstractAutowireCapableBeanFactory).createBean(String, RootBeanDefinition, Object[]) line: 498
DefaultListableBeanFactory(AbstractBeanFactory).lambda$doGetBean$0(String, RootBeanDefinition, Object[]) line: 305
2099051403.getObject() line: not available
DefaultListableBeanFactory(DefaultSingletonBeanRegistry).getSingleton(String, ObjectFactory<?>) line: 233
DefaultListableBeanFactory(AbstractBeanFactory).doGetBean(String, Class<T>, Object[], boolean) line: 303
DefaultListableBeanFactory(AbstractBeanFactory).getBean(String, Class<T>) line: 198
CommonAnnotationBeanPostProcessor.autowireResource(BeanFactory, CommonAnnotationBeanPostProcessor$LookupElement, String) line: 512
CommonAnnotationBeanPostProcessor.getResource(CommonAnnotationBeanPostProcessor$LookupElement, String) line: 483
CommonAnnotationBeanPostProcessor$ResourceElement.getResourceToInject(Object, String) line: 617
CommonAnnotationBeanPostProcessor$ResourceElement(InjectionMetadata$InjectedElement).inject(Object, String, PropertyValues) line: 172
InjectionMetadata.inject(Object, String, PropertyValues) line: 89
CommonAnnotationBeanPostProcessor.postProcessPropertyValues(PropertyValues, PropertyDescriptor[], Object, String) line: 317
DefaultListableBeanFactory(AbstractAutowireCapableBeanFactory).populateBean(String, RootBeanDefinition, BeanWrapper) line: 1353
DefaultListableBeanFactory(AbstractAutowireCapableBeanFactory).doCreateBean(String, RootBeanDefinition, Object[]) line: 579
DefaultListableBeanFactory(AbstractAutowireCapableBeanFactory).createBean(String, RootBeanDefinition, Object[]) line: 498
DefaultListableBeanFactory(AbstractBeanFactory).lambda$doGetBean$0(String, RootBeanDefinition, Object[]) line: 305
2099051403.getObject() line: not available
DefaultListableBeanFactory(DefaultSingletonBeanRegistry).getSingleton(String, ObjectFactory<?>) line: 233
DefaultListableBeanFactory(AbstractBeanFactory).doGetBean(String, Class<T>, Object[], boolean) line: 303
DefaultListableBeanFactory(AbstractBeanFactory).getBean(String) line: 193
DefaultListableBeanFactory.preInstantiateSingletons() line: 747
AnnotationConfigApplicationContext(AbstractApplicationContext).finishBeanFactoryInitialization(ConfigurableListableBeanFactory) line: 861
AnnotationConfigApplicationContext(AbstractApplicationContext).refresh() line: 542
SpringApplication.refresh(ApplicationContext) line: 750
SpringApplication.refreshContext(ConfigurableApplicationContext) line: 386
SpringApplication.run(String...) line: 327
SpringApplication.run(Class<?>[], String[]) line: 1245
SpringApplication.run(Class<?>, String...) line: 1233
Application.main(String[]) line: 51
Kafka客戶端應用程序加載spring成功后，啟動ConcurrentMessageListenerContainer監聽線程讀取客戶端offset如果是設置了事務，就開始事務執行KafkaTransactionManager.doBegin() stack
Thread [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] (Suspended (breakpoint at line 118 in KafkaTransactionManager))
KafkaTransactionManager<K,V>.doBegin(Object, TransactionDefinition) line: 118
KafkaTransactionManager<K,V>(AbstractPlatformTransactionManager).getTransaction(TransactionDefinition) line: 374
TransactionTemplate.execute(TransactionCallback<T>) line: 130
KafkaMessageListenerContainer$ListenerConsumer$1.onPartitionsAssigned(Collection<TopicPartition>) line: 434
ConsumerCoordinator.onJoinComplete(int, String, String, ByteBuffer) line: 265
ConsumerCoordinator(AbstractCoordinator).joinGroupIfNeeded() line: 363
ConsumerCoordinator(AbstractCoordinator).ensureActiveGroup() line: 310
ConsumerCoordinator.poll(long, long) line: 297
KafkaConsumer<K,V>.pollOnce(long) line: 1078
KafkaConsumer<K,V>.poll(long) line: 1043
KafkaMessageListenerContainer$ListenerConsumer.run() line: 546
Executors$RunnableAdapter<T>.call() line: 511
ListenableFutureTask<T>(FutureTask<V>).run() line: 266
Thread.run() line: 745
監聽線程讀取客戶端offset，執行producer.sendOffsetsToTransaction() stack
Thread [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] (Suspended (breakpoint at line 278 in DefaultKafkaProducerFactory$CloseSafeProducer))
DefaultKafkaProducerFactory$CloseSafeProducer<K,V>.sendOffsetsToTransaction(Map<TopicPartition,OffsetAndMetadata>, String) line: 278
KafkaMessageListenerContainer$ListenerConsumer$1$1.doInTransactionWithoutResult(TransactionStatus) line: 441
KafkaMessageListenerContainer$ListenerConsumer$1$1(TransactionCallbackWithoutResult).doInTransaction(TransactionStatus) line: 34
TransactionTemplate.execute(TransactionCallback<T>) line: 133
KafkaMessageListenerContainer$ListenerConsumer$1.onPartitionsAssigned(Collection<TopicPartition>) line: 434
ConsumerCoordinator.onJoinComplete(int, String, String, ByteBuffer) line: 265
ConsumerCoordinator(AbstractCoordinator).joinGroupIfNeeded() line: 363
ConsumerCoordinator(AbstractCoordinator).ensureActiveGroup() line: 310
ConsumerCoordinator.poll(long, long) line: 297
KafkaConsumer<K,V>.pollOnce(long) line: 1078
KafkaConsumer<K,V>.poll(long) line: 1043
KafkaMessageListenerContainer$ListenerConsumer.run() line: 546
Executors$RunnableAdapter<T>.call() line: 511
ListenableFutureTask<T>(FutureTask<V>).run() line: 266
Thread.run() line: 745
如果客戶端offset未處理則開始事務，KafkaTransactionManager.doBegin()
Thread [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] (Suspended (breakpoint at line 118 in KafkaTransactionManager))
KafkaTransactionManager<K,V>.doBegin(Object, TransactionDefinition) line: 118
KafkaTransactionManager<K,V>(AbstractPlatformTransactionManager).getTransaction(TransactionDefinition) line: 374
TransactionTemplate.execute(TransactionCallback<T>) line: 130
KafkaMessageListenerContainer$ListenerConsumer.innvokeRecordListenerInTx(ConsumerRecords<K,V>) line: 807
KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListener(ConsumerRecords<K,V>) line: 787
KafkaMessageListenerContainer$ListenerConsumer.invokeListener(ConsumerRecords<K,V>) line: 666
KafkaMessageListenerContainer$ListenerConsumer.run() line: 554
Executors$RunnableAdapter<T>.call() line: 511
ListenableFutureTask<T>(FutureTask<V>).run() line: 266
Thread.run() line: 745
調用@KafkaListener注釋方法
Thread [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] (Suspended (breakpoint at line 75 in UserServiceImpl))
UserServiceImpl.receivekafka(TransactionMessage) line: 75
UserServiceImpl$$FastClassBySpringCGLIB$$d36e166a.invoke(int, Object, Object[]) line: not available
MethodProxy.invoke(Object, Object[]) line: 204
CglibAopProxy$CglibMethodInvocation.invokeJoinpoint() line: 738
CglibAopProxy$CglibMethodInvocation(ReflectiveMethodInvocation).proceed() line: 160
680055099.proceedWithInvocation() line: not available
TransactionInterceptor(TransactionAspectSupport).invokeWithinTransaction(Method, Class<?>, InvocationCallback) line: 288
TransactionInterceptor.invoke(MethodInvocation) line: 96
CglibAopProxy$CglibMethodInvocation(ReflectiveMethodInvocation).proceed() line: 182
CglibAopProxy$DynamicAdvisedInterceptor.intercept(Object, Method, Object[], MethodProxy) line: 680
UserServiceImpl$$EnhancerBySpringCGLIB$$af9d8b47.receivekafka(TransactionMessage) line: not available
NativeMethodAccessorImpl.invoke0(Method, Object, Object[]) line: not available [native method]
NativeMethodAccessorImpl.invoke(Object, Object[]) line: 62
DelegatingMethodAccessorImpl.invoke(Object, Object[]) line: 43
Method.invoke(Object, Object...) line: 498
InvocableHandlerMethod.doInvoke(Object...) line: 181
InvocableHandlerMethod.invoke(Message<?>, Object...) line: 114
HandlerAdapter.invoke(Message<?>, Object...) line: 48
RecordMessagingMessageListenerAdapter<K,V>(MessagingMessageListenerAdapter<K,V>).invokeHandler(Object, Acknowledgment, Message<?>, Consumer<?,?>) line: 236
RecordMessagingMessageListenerAdapter<K,V>.onMessage(ConsumerRecord<K,V>, Acknowledgment, Consumer<?,?>) line: 80
RecordMessagingMessageListenerAdapter<K,V>.onMessage(Object, Acknowledgment, Consumer) line: 51
KafkaMessageListenerContainer$ListenerConsumer.doInvokeRecordListener(ConsumerRecord<K,V>, Producer) line: 856
KafkaMessageListenerContainer$ListenerConsumer.access$1700(KafkaMessageListenerContainer$ListenerConsumer, ConsumerRecord, Producer) line: 240
KafkaMessageListenerContainer$ListenerConsumer$4.doInTransactionWithoutResult(TransactionStatus) line: 816
KafkaMessageListenerContainer$ListenerConsumer$4(TransactionCallbackWithoutResult).doInTransaction(TransactionStatus) line: 34
TransactionTemplate.execute(TransactionCallback<T>) line: 133
KafkaMessageListenerContainer$ListenerConsumer.innvokeRecordListenerInTx(ConsumerRecords<K,V>) line: 807
KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListener(ConsumerRecords<K,V>) line: 787
KafkaMessageListenerContainer$ListenerConsumer.invokeListener(ConsumerRecords<K,V>) line: 666
KafkaMessageListenerContainer$ListenerConsumer.run() line: 554
Executors$RunnableAdapter<T>.call() line: 511
ListenableFutureTask<T>(FutureTask<V>).run() line: 266
Thread.run() line: 745
@KafkaListener注釋方法執行成功后執行producer.sendOffsetsToTransaction() stack
Thread [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] (Suspended (breakpoint at line 278 in DefaultKafkaProducerFactory$CloseSafeProducer))
DefaultKafkaProducerFactory$CloseSafeProducer<K,V>.sendOffsetsToTransaction(Map<TopicPartition,OffsetAndMetadata>, String) line: 278
KafkaMessageListenerContainer$ListenerConsumer.sendOffsetsToTransaction(Producer) line: 915
KafkaMessageListenerContainer$ListenerConsumer.doInvokeRecordListener(ConsumerRecord<K,V>, Producer) line: 878
KafkaMessageListenerContainer$ListenerConsumer.access$1700(KafkaMessageListenerContainer$ListenerConsumer, ConsumerRecord, Producer) line: 240
KafkaMessageListenerContainer$ListenerConsumer$4.doInTransactionWithoutResult(TransactionStatus) line: 816
KafkaMessageListenerContainer$ListenerConsumer$4(TransactionCallbackWithoutResult).doInTransaction(TransactionStatus) line: 34
TransactionTemplate.execute(TransactionCallback<T>) line: 133
KafkaMessageListenerContainer$ListenerConsumer.innvokeRecordListenerInTx(ConsumerRecords<K,V>) line: 807
KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListener(ConsumerRecords<K,V>) line: 787
KafkaMessageListenerContainer$ListenerConsumer.invokeListener(ConsumerRecords<K,V>) line: 666
KafkaMessageListenerContainer$ListenerConsumer.run() line: 554
Executors$RunnableAdapter<T>.call() line: 511
ListenableFutureTask<T>(FutureTask<V>).run() line: 266
Thread.run() line: 745