spring-batch - Spring Batch 远程分区：stepexecution 处理程序的序列化问题
问题描述
求两点帮助
- 远程分区向 Kafka 发送消息时出现异常
2020-10-18 22:43:17.106 INFO 4796 --- [ main] o.a.kafka.common.utils.AppInfoParser : Kafka version: 2.5.0
2020-10-18 22:43:17.107 INFO 4796 --- [ main] o.a.kafka.common.utils.AppInfoParser : Kafka commitId: 66563e712b0b9f84
2020-10-18 22:43:17.107 INFO 4796 --- [ main] o.a.kafka.common.utils.AppInfoParser : Kafka startTimeMs: 1603041197106
2020-10-18 22:43:17.827 INFO 4796 --- [lyContainer-C-1] o.a.k.c.c.internals.ConsumerCoordinator : [Consumer clientId=consumer-remotePartitioningConsuerGroup-1, groupId=remotePartitioningConsuerGroup] Found no committed offset for partition reply-0
2020-10-18 22:43:18.035 INFO 4796 --- [ad | producer-1] org.apache.kafka.clients.Metadata : [Producer clientId=producer-1] Cluster ID: ZBfa0qdHQIaIzmOVv1fiFg
2020-10-18 22:43:18.040 DEBUG 4796 --- [ main] o.s.i.dispatcher.UnicastingDispatcher : An exception was thrown by 'bean 'outboundGateFlow.header-enricher#1' for component 'outboundGateFlow.org.springframework.integration.config.ConsumerEndpointFactoryBean#0'; defined in: 'io.spring.batch.configuration.MasterActualConf'; from source: 'bean method outboundGateFlow'' while handling 'GenericMessage [payload=StepExecutionRequest: [jobExecutionId=112, stepExecutionId=345, stepName=workerStep], headers={sequenceNumber=1, replyChannel=bean 'org.springframework.integration.dsl.StandardIntegrationFlow#0.channel#0', correlationId=112:workerStep, id=9e635f60-9ff2-1ea1-7b11-f41ef5cb5694, sequenceSize=10, timestamp=1603041197016}]'. Failing over to the next subscriber.
org.springframework.messaging.MessageHandlingException: error occurred in message handler [bean 'outboundGateFlow.kafka:outbound-gateway#0' for component 'outboundGateFlow.org.springframework.integration.config.ConsumerEndpointFactoryBean#1'; defined in: 'io.spring.batch.configuration.MasterActualConf'; from source: 'bean method outboundGateFlow']; nested exception is org.springframework.kafka.KafkaException: Send failed; nested exception is org.apache.kafka.common.errors.SerializationException: Can't convert value of class org.springframework.batch.integration.partition.StepExecutionRequest to class org.apache.kafka.common.serialization.StringSerializer specified in value.serializer
at org.springframework.integration.support.utils.IntegrationUtils.wrapInHandlingExceptionIfNecessary(IntegrationUtils.java:192)
at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:79)
at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:115)
at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:133)
at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:106)
at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:72)
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:570)
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:520)
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:187)
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:166)
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:47)
at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:109)
at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutput(AbstractMessageProducingHandler.java:445)
at org.springframework.integration.handler.AbstractMessageProducingHandler.doProduceOutput(AbstractMessageProducingHandler.java:319)
at org.springframework.integration.handler.AbstractMessageProducingHandler.produceOutput(AbstractMessageProducingHandler.java:267)
at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutputs(AbstractMessageProducingHandler.java:231)
at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:140)
at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:62)
at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:145)
at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:106)
at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:72)
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:570)
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:520)
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:187)
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:166)
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:47)
at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:109)
at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:99)
at org.springframework.batch.integration.partition.MessageChannelPartitionHandler.handle(MessageChannelPartitionHandler.java:228)
at org.springframework.batch.core.partition.support.PartitionStep.doExecute(PartitionStep.java:106)
at org.springframework.batch.core.step.AbstractStep.execute(AbstractStep.java:208)
at org.springframework.batch.core.job.SimpleStepHandler.handleStep(SimpleStepHandler.java:148)
at org.springframework.batch.core.job.AbstractJob.handleStep(AbstractJob.java:410)
at org.springframework.batch.core.job.SimpleJob.doExecute(SimpleJob.java:136)
at org.springframework.batch.core.job.AbstractJob.execute(AbstractJob.java:319)
at org.springframework.batch.core.launch.support.SimpleJobLauncher$1.run(SimpleJobLauncher.java:147)
at org.springframework.core.task.SyncTaskExecutor.execute(SyncTaskExecutor.java:50)
at org.springframework.batch.core.launch.support.SimpleJobLauncher.run(SimpleJobLauncher.java:140)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:344)
at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:198)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163)
at org.springframework.batch.core.configuration.annotation.SimpleBatchConfiguration$PassthruAdvice.invoke(SimpleBatchConfiguration.java:127)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186)
at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:212)
at com.sun.proxy.$Proxy99.run(Unknown Source)
at org.springframework.boot.autoconfigure.batch.JobLauncherApplicationRunner.execute(JobLauncherApplicationRunner.java:199)
at org.springframework.boot.autoconfigure.batch.JobLauncherApplicationRunner.executeLocalJobs(JobLauncherApplicationRunner.java:173)
at org.springframework.boot.autoconfigure.batch.JobLauncherApplicationRunner.launchJobFromProperties(JobLauncherApplicationRunner.java:160)
at org.springframework.boot.autoconfigure.batch.JobLauncherApplicationRunner.run(JobLauncherApplicationRunner.java:155)
at org.springframework.boot.autoconfigure.batch.JobLauncherApplicationRunner.run(JobLauncherApplicationRunner.java:150)
at org.springframework.boot.SpringApplication.callRunner(SpringApplication.java:786)
at org.springframework.boot.SpringApplication.callRunners(SpringApplication.java:776)
at org.springframework.boot.SpringApplication.run(SpringApplication.java:322)
at org.springframework.boot.SpringApplication.run(SpringApplication.java:1237)
at org.springframework.boot.SpringApplication.run(SpringApplication.java:1226)
at io.spring.batch.configuration.MasterActualConf.main(MasterActualConf.java:216)
Caused by: org.springframework.kafka.KafkaException: Send failed; nested exception is org.apache.kafka.common.errors.SerializationException: Can't convert value of class org.springframework.batch.integration.partition.StepExecutionRequest to class org.apache.kafka.common.serialization.StringSerializer specified in value.serializer
at org.springframework.kafka.requestreply.ReplyingKafkaTemplate.sendAndReceive(ReplyingKafkaTemplate.java:327)
at org.springframework.kafka.requestreply.ReplyingKafkaTemplate.sendAndReceive(ReplyingKafkaTemplate.java:302)
at org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler.handleRequestMessage(KafkaProducerMessageHandler.java:399)
at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:134)
at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:62)
... 58 common frames omitted
需要知道应该使用什么序列化技术。
- 为了暂时避免上述问题，我将此消息转换为字符串，然后再将其发送到 Kafka 并获得回复超时异常。
2020-10-18 23:13:11.625 DEBUG 18892 --- [TaskScheduler-1] o.s.i.channel.PublishSubscribeChannel : postSend (sent=true) on channel 'bean 'errorChannel'', message: ErrorMessage [payload=org.springframework.messaging.MessageHandlingException: nested exception is org.springframework.kafka.requestreply.KafkaReplyTimeoutException: Reply timed out, failedMessage=GenericMessage [payload=GenericMessage [payload=StepExecutionRequest: [jobExecutionId=113, stepExecutionId=355, stepName=workerStep], headers={sequenceNumber=6, replyChannel=bean 'org.springframework.integration.dsl.StandardIntegrationFlow#0.channel#0', correlationId=113:workerStep, id=09bbd0ab-577c-d5bd-d69a-7176d6d587b9, sequenceSize=10, timestamp=1603042986620}], headers={sequenceNumber=6, replyChannel=bean 'org.springframework.integration.dsl.StandardIntegrationFlow#0.channel#0', sequenceSize=10, correlationId=113:workerStep, id=e7b7e29b-106d-a0c9-d411-ceb87c7d91c3, kafka_replyTopic=reply, timestamp=1603042986621}], headers={id=afc33702-5e97-b42e-7920-571180ce9dc1, timestamp=1603042991623}]
2020-10-18 23:13:11.639 WARN 18892 --- [TaskScheduler-1] o.s.k.r.ReplyingKafkaTemplate : Reply timed out for: ProducerRecord(topic=requests, partition=0, headers=RecordHeaders(headers = [RecordHeader(key = sequenceNumber, value = [56]), RecordHeader(key = sequenceSize, value = [49, 48]), RecordHeader(key = correlationId, value = [49, 49, 51, 58, 119, 111, 114, 107, 101, 114, 83, 116, 101, 112]), RecordHeader(key = kafka_replyTopic, value = [114, 101, 112, 108, 121]), RecordHeader(key = spring_json_header_types, value = [123, 34, 115, 101, 113, 117, 101, 110, 99, 101, 78, 117, 109, 98, 101, 114, 34, 58, 34, 106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 73, 110, 116, 101, 103, 101, 114, 34, 44, 34, 115, 101, 113, 117, 101, 110, 99, 101, 83, 105, 122, 101, 34, 58, 34, 106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 73, 110, 116, 101, 103, 101, 114, 34, 44, 34, 99, 111, 114, 114, 101, 108, 97, 116, 105, 111, 110, 73, 100, 34, 58, 34, 106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 83, 116, 114, 105, 110, 103, 34, 44, 34, 107, 97, 102, 107, 97, 95, 114, 101, 112, 108, 121, 84, 111, 112, 105, 99, 34, 58, 34, 106, 97, 118, 97, 46, 108, 97, 110, 103, 46, 83, 116, 114, 105, 110, 103, 34, 125]), RecordHeader(key = kafka_replyTopic, value = [114, 101, 112, 108, 121]), RecordHeader(key = kafka_correlationId, value = [-33, 26, 119, 80, -98, 58, 67, 14, -118, -126, 41, -19, -84, -35, -48, 32])], isReadOnly = true), key=null, value=GenericMessage [payload=StepExecutionRequest: [jobExecutionId=113, stepExecutionId=354, stepName=workerStep], headers={sequenceNumber=8, replyChannel=bean 'org.springframework.integration.dsl.StandardIntegrationFlow#0.channel#0', correlationId=113:workerStep, id=87938774-9e2d-05aa-d9a0-71672c2e2f90, sequenceSize=10, timestamp=1603042986634}], timestamp=null) with correlationId: [-43727104148615416681456858404470534112]
2020-10-18 23:13:11.639 DEBUG 18892 --- [TaskScheduler-1] o.s.i.channel.PublishSubscribeChannel : preSend on channel 'bean 'errorChannel'', message: ErrorMessage [payload=org.springframework.messaging.MessageHandlingException: nested exception is org.springframework.kafka.requestreply.KafkaReplyTimeoutException: Reply timed out, failedMessage=GenericMessage [payload=GenericMessage [payload=StepExecutionRequest: [jobExecutionId=113, stepExecutionId=354, stepName=workerStep], headers={sequenceNumber=8, replyChannel=bean 'org.springframework.integration.dsl.StandardIntegrationFlow#0.channel#0', correlationId=113:workerStep, id=87938774-9e2d-05aa-d9a0-71672c2e2f90, sequenceSize=10, timestamp=1603042986634}], headers={sequenceNumber=8, replyChannel=bean 'org.springframework.integration.dsl.StandardIntegrationFlow#0.channel#0', sequenceSize=10, correlationId=113:workerStep, id=77d7edd9-0bec-c517-ef5d-90687bed0434, kafka_replyTopic=reply, timestamp=1603042986637}], headers={id=83e1628f-d4d6-14a6-e052-5b3eee0139d5, timestamp=1603042991639}]
2020-10-18 23:13:11.639 DEBUG 18892 --- [TaskScheduler-1] o.s.integration.handler.LoggingHandler : bean '_org.springframework.integration.errorLogger.handler' for component '_org.springframework.integration.errorLogger' received message: ErrorMessage [payload=org.springframework.messaging.MessageHandlingException: nested exception is org.springframework.kafka.requestreply.KafkaReplyTimeoutException: Reply timed out, failedMessage=GenericMessage [payload=GenericMessage [payload=StepExecutionRequest: [jobExecutionId=113, stepExecutionId=354, stepName=workerStep], headers={sequenceNumber=8, replyChannel=bean 'org.springframework.integration.dsl.StandardIntegrationFlow#0.channel#0', correlationId=113:workerStep, id=87938774-9e2d-05aa-d9a0-71672c2e2f90, sequenceSize=10, timestamp=1603042986634}], headers={sequenceNumber=8, replyChannel=bean 'org.springframework.integration.dsl.StandardIntegrationFlow#0.channel#0', sequenceSize=10, correlationId=113:workerStep, id=77d7edd9-0bec-c517-ef5d-90687bed0434, kafka_replyTopic=reply, timestamp=1603042986637}], headers={id=83e1628f-d4d6-14a6-e052-5b3eee0139d5, timestamp=1603042991639}]
2020-10-18 23:13:11.640 ERROR 18892 --- [TaskScheduler-1] o.s.integration.handler.LoggingHandler : org.springframework.messaging.MessageHandlingException: nested exception is org.springframework.kafka.requestreply.KafkaReplyTimeoutException: Reply timed out, failedMessage=GenericMessage [payload=GenericMessage [payload=StepExecutionRequest: [jobExecutionId=113, stepExecutionId=354, stepName=workerStep], headers={sequenceNumber=8, replyChannel=bean 'org.springframework.integration.dsl.StandardIntegrationFlow#0.channel#0', correlationId=113:workerStep, id=87938774-9e2d-05aa-d9a0-71672c2e2f90, sequenceSize=10, timestamp=1603042986634}], headers={sequenceNumber=8, replyChannel=bean 'org.springframework.integration.dsl.StandardIntegrationFlow#0.channel#0', sequenceSize=10, correlationId=113:workerStep, id=77d7edd9-0bec-c517-ef5d-90687bed0434, kafka_replyTopic=reply, timestamp=1603042986637}]
at org.springframework.integration.handler.AbstractMessageProducingHandler.sendErrorMessage(AbstractMessageProducingHandler.java:476)
at org.springframework.integration.handler.AbstractMessageProducingHandler$ReplyFutureCallback.onFailure(AbstractMessageProducingHandler.java:543)
at org.springframework.util.concurrent.ListenableFutureCallbackRegistry.notifyFailure(ListenableFutureCallbackRegistry.java:86)
at org.springframework.util.concurrent.ListenableFutureCallbackRegistry.failure(ListenableFutureCallbackRegistry.java:158)
at org.springframework.util.concurrent.ListenableFutureTask.done(ListenableFutureTask.java:100)
at org.springframework.util.concurrent.SettableListenableFuture$SettableTask.done(SettableListenableFuture.java:175)
at java.util.concurrent.FutureTask.finishCompletion(FutureTask.java:384)
at java.util.concurrent.FutureTask.setException(FutureTask.java:251)
at org.springframework.util.concurrent.SettableListenableFuture$SettableTask.setExceptionResult(SettableListenableFuture.java:163)
at org.springframework.util.concurrent.SettableListenableFuture.setException(SettableListenableFuture.java:70)
at org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler$ConvertingReplyFuture$1.onFailure(KafkaProducerMessageHandler.java:627)
at org.springframework.util.concurrent.ListenableFutureCallbackRegistry.notifyFailure(ListenableFutureCallbackRegistry.java:86)
at org.springframework.util.concurrent.ListenableFutureCallbackRegistry.failure(ListenableFutureCallbackRegistry.java:158)
at org.springframework.util.concurrent.ListenableFutureTask.done(ListenableFutureTask.java:100)
at org.springframework.util.concurrent.SettableListenableFuture$SettableTask.done(SettableListenableFuture.java:175)
at java.util.concurrent.FutureTask.finishCompletion(FutureTask.java:384)
at java.util.concurrent.FutureTask.setException(FutureTask.java:251)
at org.springframework.util.concurrent.SettableListenableFuture$SettableTask.setExceptionResult(SettableListenableFuture.java:163)
at org.springframework.util.concurrent.SettableListenableFuture.setException(SettableListenableFuture.java:70)
at org.springframework.kafka.requestreply.ReplyingKafkaTemplate.lambda$scheduleTimeout$3(ReplyingKafkaTemplate.java:339)
at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Caused by: org.springframework.kafka.requestreply.KafkaReplyTimeoutException: Reply timed out
... 9 more
====================================== 转换为字符串后，我发现的消息是
payload GenericMessage [payload=StepExecutionRequest: [jobExecutionId=113, stepExecutionId=354, stepName=workerStep], headers={sequenceNumber=8, replyChannel=bean 'org.springframework.integration.dsl.StandardIntegrationFlow#0.channel#0', correlationId=113:workerStep, id=87938774-9e2d-05aa-d9a0-71672c2e2f90, sequenceSize=10, timestamp=1603042986634}]
想知道 master 和 slave 将通过 Kafka 消息进行通信，或者他们将使用存储库 DB 共享信息（我想知道，如果我将 repo 保留在节点本地，它会起作用吗？否则 master 和 slave 将需要共享存储库）。
解决方案
您需要使用JsonSerializer生产者值序列化器和JsonDeserializer消费者值反序列化器，而不是简单的字符串序列化器/反序列化器。
推荐阅读
- r - 在 R 闪亮的控制台中不生成输出
- markdown - Zeppelin - 动态表单 - 如何创建用于输入密码的屏蔽字段
- laravel - 如何在 laravel 中处理动态 url
- html - 我以百分比 (%) 为 div 元素设置“高度”CSS 属性，但 html 不呈现任何内容。谁能帮我？
- sas - %QUOTE 和 %BQUOTE 之间的区别
- sql-server-2012 - 将聚簇索引扫描优化为聚簇索引查找
- python - 组合约束满足和优化
- flask - 是否可以在 Flask Admin 的编辑视图中使用 column_formatters？
- c++ - 删除 XML 文件 C++ 中的子节点
- javascript - 获取快速记忆的结果