Skip to content

Commit d68a154

Browse files
author
pierre lemee
committed
Upgraded flume-ng-core version to 1.6.0-cdh5.5.0
1 parent 26b7ec3 commit d68a154

File tree

2 files changed

+100
-104
lines changed

2 files changed

+100
-104
lines changed

‎pom.xml

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,7 @@
99
<url>http://maven.apache.org</url>
1010
<properties>
1111
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
12-
<flume.version>1.2.0-cdh4.1.2</flume.version>
12+
<flume.version>1.6.0-cdh5.5.0</flume.version>
1313
<rabbitmq.version>2.8.2</rabbitmq.version>
1414
<junit.version>4.10</junit.version>
1515
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>

‎src/main/java/org/apache/flume/source/rabbitmq/RabbitMQSource.java

Lines changed: 99 additions & 103 deletions
Original file line numberDiff line numberDiff line change
@@ -22,16 +22,9 @@
2222
import java.util.Map;
2323

2424
import org.apache.commons.lang.StringUtils;
25-
import org.apache.flume.Context;
26-
import org.apache.flume.CounterGroup;
27-
import org.apache.flume.Event;
28-
import org.apache.flume.EventDeliveryException;
29-
import org.apache.flume.PollableSource;
30-
import org.apache.flume.RabbitMQConstants;
31-
import org.apache.flume.RabbitMQUtil;
32-
import org.apache.flume.conf.Configurable;
25+
import org.apache.flume.*;
3326
import org.apache.flume.event.SimpleEvent;
34-
import org.apache.flume.source.AbstractSource;
27+
import org.apache.flume.source.AbstractPollableSource;
3528
import org.slf4j.Logger;
3629
import org.slf4j.LoggerFactory;
3730

@@ -41,7 +34,7 @@
4134
import com.rabbitmq.client.QueueingConsumer;
4235

4336

44-
public class RabbitMQSource extends AbstractSource implements Configurable, PollableSource {
37+
public class RabbitMQSource extends AbstractPollableSource {
4538
private static final Logger log = LoggerFactory.getLogger(RabbitMQSource.class);
4639
private CounterGroup _CounterGroup;
4740
private ConnectionFactory _ConnectionFactory;
@@ -55,48 +48,22 @@ public class RabbitMQSource extends AbstractSource implements Configurable, Poll
5548
public RabbitMQSource(){
5649
_CounterGroup = new CounterGroup();
5750
}
58-
59-
60-
@Override
61-
public void configure(Context context) {
62-
_ConnectionFactory = RabbitMQUtil.getFactory(context);
63-
_QueueName = RabbitMQUtil.getQueueName(context);
64-
_ExchangeName = RabbitMQUtil.getExchangeName(context);
65-
_Topics = RabbitMQUtil.getTopics(context);
66-
67-
ensureConfigCompleteness( context );
68-
}
69-
70-
@Override
71-
public synchronized void stop() {
72-
RabbitMQUtil.close(_Connection, _Channel);
73-
super.stop();
74-
}
75-
76-
private void resetConnection(){
77-
_CounterGroup.incrementAndGet(RabbitMQConstants.COUNTER_EXCEPTION);
78-
if(log.isWarnEnabled())log.warn(this.getName() + " - Closing RabbitMQ connection and channel due to exception.");
79-
RabbitMQUtil.close(_Connection, _Channel);
80-
_Connection=null;
81-
_Channel=null;
82-
_Consumer=null;
83-
}
84-
51+
8552
@Override
86-
public PollableSource.Status process() throws EventDeliveryException {
53+
protected Status doProcess() throws EventDeliveryException {
8754
if(null==_Connection){
8855
try {
8956
if(log.isInfoEnabled())log.info(this.getName() + " - Opening connection to " + _ConnectionFactory.getHost() + ":" + _ConnectionFactory.getPort());
9057
_Connection = _ConnectionFactory.newConnection();
91-
_CounterGroup.incrementAndGet(RabbitMQConstants.COUNTER_NEW_CONNECTION);
58+
_CounterGroup.incrementAndGet(RabbitMQConstants.COUNTER_NEW_CONNECTION);
9259
_Channel = null;
9360
} catch(Exception ex) {
9461
if(log.isErrorEnabled()) log.error(this.getName() + " - Exception while establishing connection.", ex);
9562
resetConnection();
9663
return Status.BACKOFF;
97-
}
64+
}
9865
}
99-
66+
10067
if(null==_Channel){
10168
try {
10269
if(log.isInfoEnabled())log.info(this.getName() + " - creating channel...");
@@ -105,96 +72,125 @@ public PollableSource.Status process() throws EventDeliveryException {
10572
if(log.isInfoEnabled())log.info(this.getName() + " - Connected to " + _ConnectionFactory.getHost() + ":" + _ConnectionFactory.getPort());
10673
_Consumer=null;
10774
if( StringUtils.isNotEmpty(_ExchangeName) ) {
108-
try {
109-
//declare an exchange
110-
_Channel.exchangeDeclarePassive(_ExchangeName);
111-
112-
//only grab a default queuename if one is not specified in config
113-
if( StringUtils.isEmpty( _QueueName ) ) {
114-
_QueueName = _Channel.queueDeclare().getQueue();
115-
}
116-
117-
//for each topic, bind to the key
118-
if( null != _Topics ) {
119-
for ( String topic : _Topics ) {
120-
_Channel.queueBind(_QueueName, _ExchangeName, topic);
121-
}
122-
}
123-
}
124-
catch( Exception ex ) {
75+
try {
76+
//declare an exchange
77+
_Channel.exchangeDeclarePassive(_ExchangeName);
78+
79+
//only grab a default queuename if one is not specified in config
80+
if( StringUtils.isEmpty( _QueueName ) ) {
81+
_QueueName = _Channel.queueDeclare().getQueue();
82+
}
83+
84+
//for each topic, bind to the key
85+
if( null != _Topics ) {
86+
for ( String topic : _Topics ) {
87+
_Channel.queueBind(_QueueName, _ExchangeName, topic);
88+
}
89+
}
90+
}
91+
catch( Exception ex ) {
12592
if(log.isErrorEnabled()) log.error(this.getName() + " - Exception while declaring exchange.", ex);
12693
resetConnection();
12794
return Status.BACKOFF;
128-
}
95+
}
12996
}
130-
} catch(Exception ex) {
97+
} catch(Exception ex) {
13198
if(log.isErrorEnabled()) log.error(this.getName() + " - Exception while creating channel.", ex);
13299
resetConnection();
133100
return Status.BACKOFF;
134-
}
101+
}
135102
}
136103
if(null == _Consumer){
137-
try{
138-
_Consumer = new QueueingConsumer(_Channel);
139-
_Channel.basicConsume(_QueueName, false, _Consumer);
140-
}catch( Exception ex ) {
104+
try{
105+
_Consumer = new QueueingConsumer(_Channel);
106+
_Channel.basicConsume(_QueueName, false, _Consumer);
107+
}catch( Exception ex ) {
141108
if(log.isErrorEnabled()) log.error(this.getName() + " - Exception while registering consumer", ex);
142109
resetConnection();
143110
return Status.BACKOFF;
144-
}
111+
}
112+
}
113+
114+
QueueingConsumer.Delivery delivery;
115+
116+
try {
117+
delivery = _Consumer.nextDelivery();
118+
_CounterGroup.incrementAndGet(RabbitMQConstants.COUNTER_GET);
119+
}
120+
catch (Exception ex) {
121+
_CounterGroup.incrementAndGet(RabbitMQConstants.COUNTER_EXCEPTION);
122+
if (log.isErrorEnabled())
123+
log.error(this.getName() + " - Exception thrown while pulling from queue.", ex);
124+
resetConnection();
125+
return Status.BACKOFF;
126+
}
127+
128+
if (null == delivery) {
129+
_CounterGroup.incrementAndGet(RabbitMQConstants.COUNTER_GET_MISS);
130+
return Status.BACKOFF;
145131
}
146132

147-
QueueingConsumer.Delivery delivery;
148-
149-
try {
150-
delivery = _Consumer.nextDelivery();
151-
_CounterGroup.incrementAndGet(RabbitMQConstants.COUNTER_GET);
152-
}
153-
catch (Exception ex) {
154-
_CounterGroup.incrementAndGet(RabbitMQConstants.COUNTER_EXCEPTION);
155-
if (log.isErrorEnabled())
156-
log.error(this.getName() + " - Exception thrown while pulling from queue.", ex);
157-
resetConnection();
158-
return Status.BACKOFF;
159-
}
160-
161-
if (null == delivery) {
162-
_CounterGroup.incrementAndGet(RabbitMQConstants.COUNTER_GET_MISS);
163-
return Status.BACKOFF;
164-
}
165-
166-
try {
167-
Map<String, String> properties = RabbitMQUtil.getHeaders(delivery.getProperties());
168-
Event event = new SimpleEvent();
169-
event.setBody(delivery.getBody());
170-
event.setHeaders(properties);
171-
172-
getChannelProcessor().processEvent(event);
173-
} catch (Exception ex) {
174-
if (log.isErrorEnabled())
175-
log.error(this.getName() + " - Exception thrown while processing event", ex);
176-
177-
return Status.BACKOFF;
178-
}
133+
try {
134+
Map<String, String> properties = RabbitMQUtil.getHeaders(delivery.getProperties());
135+
Event event = new SimpleEvent();
136+
event.setBody(delivery.getBody());
137+
event.setHeaders(properties);
138+
139+
getChannelProcessor().processEvent(event);
140+
} catch (Exception ex) {
141+
if (log.isErrorEnabled())
142+
log.error(this.getName() + " - Exception thrown while processing event", ex);
143+
144+
return Status.BACKOFF;
145+
}
179146

180147
try {
181-
_Channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
148+
_Channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
182149
_CounterGroup.incrementAndGet(RabbitMQConstants.COUNTER_ACK);
183150
} catch(Exception ex){
184151
_CounterGroup.incrementAndGet(RabbitMQConstants.COUNTER_EXCEPTION);
185152
if(log.isErrorEnabled())log.error(this.getName() + " - Exception thrown while sending ack to queue", ex);
186153
resetConnection();
187-
return Status.BACKOFF;
154+
return Status.BACKOFF;
188155
}
189-
190-
return Status.READY;
156+
157+
return Status.READY;
158+
}
159+
160+
@Override
161+
protected void doConfigure(Context context) throws FlumeException {
162+
_ConnectionFactory = RabbitMQUtil.getFactory(context);
163+
_QueueName = RabbitMQUtil.getQueueName(context);
164+
_ExchangeName = RabbitMQUtil.getExchangeName(context);
165+
_Topics = RabbitMQUtil.getTopics(context);
166+
167+
ensureConfigCompleteness( context );
168+
}
169+
170+
@Override
171+
protected void doStart() throws FlumeException {
172+
173+
}
174+
175+
@Override
176+
protected void doStop() throws FlumeException {
177+
RabbitMQUtil.close(_Connection, _Channel);
178+
super.stop();
191179
}
192180

181+
private void resetConnection(){
182+
_CounterGroup.incrementAndGet(RabbitMQConstants.COUNTER_EXCEPTION);
183+
if(log.isWarnEnabled())log.warn(this.getName() + " - Closing RabbitMQ connection and channel due to exception.");
184+
RabbitMQUtil.close(_Connection, _Channel);
185+
_Connection=null;
186+
_Channel=null;
187+
_Consumer=null;
188+
}
193189

194190
/**
195191
* Verify that the required configuration is set
196192
*
197-
* @param context
193+
* @param context Context
198194
*/
199195
private void ensureConfigCompleteness( Context context ) {
200196

0 commit comments

Comments
 (0)