rabbitmq-flume-plugin

A Flume plugin that provides a RabbitMQ Source and Sink. While there are other Flume(ng) plugins that do this as well, this implementation aims to be highly performant and provide tools for mapping message properties to Flume event headers.

This plugin was developed and tested against Flume 1.5.0.

Download

You can find the latest and greatest version on the Releases page.

Installation Instructions

To install the plugin, copy the jar file to the $FLUME_LIB directory. For example, the $FLUME_LIB directory for Cloudera (CDH5) installed Flume, the $FLUME_LIB is /usr/lib/flume/lib.

Behavior

Source

The RabbitMQ Source component enables RabbitMQ to act as a source for Flume events. You must create and bind the queue you would like the source to consume from manually in RabbitMQ, prior to starting Flume.

When the Source consumes a message, any message properties() that are set on the message will be added to the Flume Event headers, including the exchange and routing key for the message. () This excludes the headers property which is a free-form key/value table. This is currently excluded due to the complexity of correctly parsing and dealing with the different data types that could be returned as values in the headers table.

Using the Flume Event headers, you could use the type property, exchange, or routing-key as part of the file path or name when using the HDFS sink.

By default, there is a single consumer thread in the RabbitMQ source.

Configuration

Each component has its own configuration directives, but the Source and Sink share common RabbitMQ connection parameters.

Source

The Source component has the following configuration options:

Possible event header keys

• exchange
• routing-key
• app-id
• content-encoding
• content-type
• correlation-id
• delivery-mode
• expires
• message-id
• priority
• reply-to
• timestamp
• type
• user-id

Example

a1.sources.r1.channels = c1
a1.sources.r1.type = com.aweber.flume.source.rabbitmq.RabbitMQSource
a1.sources.r1.host = localhost
a1.sources.r1.port = 5672
a1.sources.r1.virtual-host = /
a1.sources.r1.username = flume
a1.sources.r1.password = rabbitmq
a1.sources.r1.queue = events_for_s3
a1.sources.r1.prefetchCount = 10

Sink

The RabbitMQ Sink allows for Flume events to be published to RabbitMQ.

Headers

When publishing an event message, the RabbitMQ Sink will first look to the event headers for a routing-key entry. If it is set, it will use that value when publishing the message. If it is not set, it will fall back to the configured routing-key value which defaults to an empty string.

If the auto-properties configuration option is enabled (default), the event headers will be examined for standard AMQP Basic.Properties entries (sans the headers AMQP table). If the property is set in the event headers, it will be set in the message properties. Additionally, if the app-id value is not set in the headers, it will default to RabbitMQSink. If timestamp is not set in the headers, it will default to the current system time.

Available property keys

• app-id
• content-encoding
• content-type
• correlation-id
• delivery-mode
• expires
• message-id
• priority
• reply-to
• timestamp
• type
• user-id

Example

a1.sinks.k1.channels = c1
a1.sinks.k1.type = com.aweber.flume.sink.rabbitmq.RabbitMQSink
a1.sinks.k1.host = localhost
a1.sinks.k1.port = 5672
a1.sinks.k1.virtual-host = /
a1.sinks.k1.username = flume
a1.sinks.k1.password = rabbitmq
a1.sinks.k1.exchange = amq.topic
a1.sinks.k1.routing-key = flume.event
a1.sinks.k1.publisher-confirms = true

Version History

• v1.0.x - Not yet released
  • Timeout for consumer to allow clean shutdown and configuration reload
  • Requeuing / discarding of failed messages by broker now configurable (before: hardcoded to discarding)
• v1.0.2 - Released 2015-01-12
  • Fix an issue where the counter names were null for both Source and Sink

Build Instructions

To build from source, use maven:

mvn package

This will download all of the dependencies required for building the plugin and provide a jar file in the target/ directory.