I am trying to figure out why my protocol buffer payload is being incorrectly de-serialized when received from a Spring AMQP listener in my Java app. The protocol buffer is written in a Go service and sent to a Rabbit MQ instance which is then picked up by the Java Spring app and de-serialized into an object instance.
For some reason when received the marshalled object has field 2 with the value of field 1 while field 1 is null in the resulting object.
Additionally the repeated network_interfaces field with key 5 is null on the object.
Interestingly if I use the debugger when picking up a message from the queue, the missing fields are visible and contain the correct data on the instantiated protobuf object under unknownFields so it looks like a case of the proto parser not working correctly.
The debugger shows that the field keys are off, i.e. field 1 has the key 2 and so on.
So far I have:
- Verified the proto files are the same in both services, the Spring app is using the proto file from the other repo by importing it as a git submodule
- Attempted to use a
Spring.messaging.amqpMessageConverter to deserialize the proto explicitly based on message headers with no success - Tried different message content type and encoding headers when sending the message, these don't appear to make a difference as I am only using the message body.
- Verified that if the Go app listens to the queue that it can deserialize and unmarshal a message it sent correctly, so this has to be a Spring config issue or similar.
The Proto3 message definition in question:
message HostStateMessage {
message NetworkInterface {
string name = 1;
string mac_address = 2;
string ipv4 = 3;
string ipv6 = 4;
}
string hostname = 1;
string description = 2;
HostType host_type = 3;
repeated NetworkInterface network_interfaces = 4;
}
The Go function which writes the protocol buffer to RabbitMQ as a message:
func PublishHostMessage(config *AMQPQueueConfiguration, conn *amqp.Connection, message *protos.HostStateMessage) error {
channel, err := conn.Channel()
if err != nil {
fmt.Errorf("Creating channel failed: %s\n", err)
return err
}
queue, err := channel.QueueDeclare(config.HostQueueName, true, false, false, false, nil)
if err != nil {
fmt.Errorf("Failed to declare queue: %s\n", err)
return err
}
serialMsg, err := proto.Marshal(message)
if err != nil {
fmt.Errorf("Failed to serialize proto message: %s\n", err)
return err
}
headers := amqp.Table{}
headers["messageType"] = "HostStateMessage"
fmt.Println(string(serialMsg))
err = channel.Publish("", queue.Name, false, false, amqp.Publishing{Headers: headers, ContentType: "application/x-protobuf", Body: serialMsg})
if err != nil {
fmt.Errorf("Failed to send message: %s\n", err)
return err
}
fmt.Printf("Sent host message at: %s\n", time.Now())
return nil
}
The MessageReceiver.java Class:
@Service
public class MessageReceiver {
@Autowired RabbitConfiguration rabbitConfiguration;
@Autowired HostDao hostDao;
@RabbitListener(queues = "#{rabbitConfiguration.hostQueueName}")
public void consumeHostNotification(Message in) {
try {
HostStateMessage message = HostStateMessage.parseFrom(in.getBody());
hostDao.addOrUpdateHostStateFromMessage(message);
} catch (InvalidProtocolBufferException e) {
System.out.println(e.getStackTrace());
}
}
}
The Spring AMQP @Configuration bean:
@Configuration
@EnableAutoConfiguration
public class RabbitConfiguration {
@Value("${rabbitmq.hostname}")
private String queueHost;
@Value("${rabbitmq.port}")
private int queuePort;
@Value("${rabbitmq.queues.host-queue-name}")
public String hostQueueName;
@Bean
public CachingConnectionFactory connectionFactory() {
return new CachingConnectionFactory(queueHost, queuePort);
}
@Bean
public Queue hostNotificationQueue() {
return new Queue(hostQueueName);
}
}
Go is using google.golang.org/protobuf v1.27.1 in Go modules.
Java is using com.google.protobuf 3.19.4 as a Maven dependency. Along with protobuf-maven-plugin 0.6.1 to do the compilation with protoc.
This is a really confusing issue, would be great to get some insight.