Fork me on GitHub

AMQP Sink Connector Architecture Notes

Aiven sink connector for AMQP on Apache Kafka connects to an AMQP provider using the protonj2 client from the Apache QPID™ messaging library to write AMQP messages. It utilizes the Aiven Kafka Connector Framework to do most of the heavy lifting.

The sink processor uses a write strategy to write the Kafka messages into the AMQP message format and deliver them to the AMQP system.

As of this writing the AMQP sink connector supports two strategies:

The raw strategy assumes that the entire AMQP message was serialized into the Kafka message value as a byte[]. The format for this buffer is found in the AmqpConverter code in the connectors-common-for-amqp package. This strategy is intended for use with the Aiven AMQP source connector RAW format and the AmqpConverter. If messages are placed into the Kafka system using Aiven AMQP source connector and the RAW format and the AMQPConverter is used for the value converter then the Aiven AMQP sink connector using the RAW format and RAW strategy will place the same message on the AMQP queue. All properties and annotations for the message are found in the message as serialzed into the Kafka value and are not impacted by changes to the amqp. prefixed Kafka headers. There are some changes to the data elements for the receiving AMQP system.

When the messages placed on AMQP queue by the sink processor in RAW strategy are read by another AMQP client the following data items will have been changed.

  1. Delivery count - The delivery count on the receiver will reflect the delivery count since the message was placed on the queue by the sink process. The original message count will be lost to the remote. However, the original delivery count is available in the sink processor as the amqp.deliveryCount Kafka message header and may be used in an SMT to preserve the value.
  2. First acquirer - The first acquirer recorded on the receiver will reflect the first acquirer since the message was placed on the queue by the sink process. The original first acquirer will be lost to the remote. However, the original first acquirer is available in the sink processor as the amqp.firstAcquirer Kafka message header and may be used in an SMT to preserve the value.

The body strategy assumes that the Kafka value object should be placed in the AMQP body field. All other settings are set based on the presence of kafka headers that have keys that start with amqp.. This strategy is useful when the original source is not the Aiven AMQP Sourxe connector. The additional header information should be placed in the Kafka headers. This may be done when the original Kafka message is created or though the uses of SMTs. There are some changes to the data elements for the receiving AMQP system.

  1. Delivery count - The delivery count on the receiver will reflect the delivery count since the message was placed on the queue by the sink process. The original message count will be lost to the remote. However, the original delivery count is available in the sink processor as the amqp.deliveryCount Kafka message header and may be used in an SMT to preserve the value.
  2. First acquirer - The first acquirer recorded on the receiver will reflect the first acquirer since the message was placed on the queue by the sink process. The original first acquirer will be lost to the remote. However, the original first acquirer is available in the sink processor as the amqp.firstAcquirer Kafka message header and may be used in an SMT to preserve the value.
  3. Priority - The priority of the message will be set by the priority defined in the AMQP system.
  4. Time to live (TTL) - The time to live will be reset to and based on the time the new message is delivered to the AMQP system.
  5. User Id - The user id will be changed to the ID that the the sink connector uses to communicate with the AMQP system. The original user id will be lost to the remote. However, the original user id is available in the sink processor as the amqp.userId Kafka message header and may be used in an SMT to preserve the value.

Data Mapping

The Headers generated from the message has a number of properties. If the property listed below as not set in the message it will not be present. The potential properties are:

  • absoluteExpiry : The absolute time for the expiration of the message.
  • annotations : Conceptually a map of strings to objects. The JSON serialization of this is a standard JSON map serialization.
  • contentEncoding : The content-encoding property is used as a modifier to the content-type. When present, its value indicates what additional content encodings have been applied to the application-data, and thus what decoding mechanisms need to be applied in order to obtain the media-type referenced by the content-type header field.
  • contentType : The RFC-2046 MIME type for the message's application-data section (body). As per RFC-2046 this can contain a charset parameter defining the character encoding used: e.g., ‘text/plain; charset=“utf-8”’.
  • correlationId : An id that correlates messages together.
  • creationTime : the absolute time for the creation of the message.
  • deliveryCount : the number of failed delivery attempts this message has been part of.
  • durable : the flag that indicates if the message is durable.
  • firstAcquirer : if this message has been acquired by another link previously.
  • footers : conceptually a map of strings to objects. The JSON serialization of this is a standard JSON map serialization.
  • groupId : the group id for the message.
  • groupSequence : the group sequence number for the message.
  • messageId : The message Id if any was provided.
  • properties : conceptually a map of strings to objects. The JSON serialization of this is a standard JSON map serialization.
  • replyTo : the suggested reply destination.
  • replyToGroup : the group to which replies should be sent.
  • subject : the subject of the message
  • to : The intended destination of the message if set.
  • userId : The Id of the user that created the message if it was provided.