У меня есть задание Dataflow, которое записывает авро-сообщения в PubSub:
PubsubIO.writeAvros(Session.class).to(sessionTopic)
Затем я хочу обработать сообщения в приложении Spring. В общем, я использую org.springframework.cloud.spring-cloud-gcp-starter-pubsub и org.springframework.cloud.spring-cloud-stream-schema зависимости.
@Bean
public PubSubInboundChannelAdapter messageChannelAdapter(
@Qualifier("sessionInputChannel") MessageChannel inputChannel,
PubSubOperations pubSubTemplate) {
PubSubInboundChannelAdapter adapter = new PubSubInboundChannelAdapter(pubSubTemplate,
"sessions-subscription");
adapter.setOutputChannel(inputChannel);
adapter.setMessageConverter(sessionMessageConverter());
adapter.setAckMode(AckMode.MANUAL);
return adapter;
}
private MessageConverter sessionMessageConverter() {
AvroSchemaMessageConverter converter = new AvroSchemaMessageConverter();
converter.setSchema(Session.getClassSchema());
return converter;
}
@Bean
public MessageChannel sessionInputChannel() {
return new DirectChannel();
}
@Bean
@ServiceActivator(inputChannel = "sessionInputChannel")
public MessageHandler messageReceiver() {
return message -> {
log.info("Message arrived! Payload: " + message.getPayload());
Session session = (Session) message.getPayload();
log.info("Session: " + session);
AckReplyConsumer consumer = (AckReplyConsumer) message.getHeaders()
.get(GcpPubSubHeaders.ACKNOWLEDGEMENT);
consumer.ack();
};
}
К сожалению, я получаю следующее исключение:
java.lang.NullPointerException: in com.example.model.Session null of com.example.model.Session
at org.apache.avro.reflect.ReflectDatumWriter.write(ReflectDatumWriter.java:161) ~[avro-1.8.2.jar:1.8.2]
at org.apache.avro.generic.GenericDatumWriter.write(GenericDatumWriter.java:62) ~[avro-1.8.2.jar:1.8.2]
at org.springframework.cloud.stream.schema.avro.AbstractAvroMessageConverter.convertToInternal(AbstractAvroMessageConverter.java:183) ~[spring-cloud-stream-schema-2.0.0.RELEASE.jar:2.0.0.RELEASE]
at org.springframework.messaging.converter.AbstractMessageConverter.toMessage(AbstractMessageConverter.java:201) ~[spring-messaging-5.0.6.RELEASE.jar:5.0.6.RELEASE]
at org.springframework.messaging.converter.AbstractMessageConverter.toMessage(AbstractMessageConverter.java:191) ~[spring-messaging-5.0.6.RELEASE.jar:5.0.6.RELEASE]
at org.springframework.cloud.gcp.pubsub.integration.inbound.PubSubInboundChannelAdapter.receiveMessage(PubSubInboundChannelAdapter.java:88) ~[spring-cloud-gcp-pubsub-1.0.0.M3.jar:1.0.0.M3]
at com.google.cloud.pubsub.v1.MessageDispatcher$4.run(MessageDispatcher.java:405) ~[google-cloud-pubsub-0.43.0-beta.jar:0.43.0-beta]
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [na:1.8.0_171]
at java.util.concurrent.FutureTask.run(FutureTask.java:266) [na:1.8.0_171]
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) [na:1.8.0_171]
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) [na:1.8.0_171]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [na:1.8.0_171]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [na:1.8.0_171]
at java.lang.Thread.run(Thread.java:748) [na:1.8.0_171]
Caused by: java.lang.NullPointerException: null
at org.apache.avro.reflect.ReflectData.getField(ReflectData.java:158) ~[avro-1.8.2.jar:1.8.2]
at org.apache.avro.generic.GenericDatumWriter.writeField(GenericDatumWriter.java:164) ~[avro-1.8.2.jar:1.8.2]
at org.apache.avro.specific.SpecificDatumWriter.writeField(SpecificDatumWriter.java:90) ~[avro-1.8.2.jar:1.8.2]
at org.apache.avro.reflect.ReflectDatumWriter.writeField(ReflectDatumWriter.java:191) ~[avro-1.8.2.jar:1.8.2]
at org.apache.avro.generic.GenericDatumWriter.writeRecord(GenericDatumWriter.java:156) ~[avro-1.8.2.jar:1.8.2]
at org.apache.avro.generic.GenericDatumWriter.writeWithoutConversion(GenericDatumWriter.java:118) ~[avro-1.8.2.jar:1.8.2]
at org.apache.avro.generic.GenericDatumWriter.write(GenericDatumWriter.java:75) ~[avro-1.8.2.jar:1.8.2]
at org.apache.avro.reflect.ReflectDatumWriter.write(ReflectDatumWriter.java:159) ~[avro-1.8.2.jar:1.8.2]
Может кто-нибудь сказать мне, как использовать сообщения авро? Что я делаю не так?