This is an automated email from the ASF dual-hosted git repository.
ijokarumawak pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/master by this push:
new cded30b NIFI-5953 Manage GetTwitter connection retries on '420
Enhance Your Calm' exceptions
cded30b is described below
commit cded30b3d2dc0497cf3d06051f55ca304c744250
Author: Kourge <[email protected]>
AuthorDate: Mon Jan 28 14:10:54 2019 +0100
NIFI-5953 Manage GetTwitter connection retries on '420 Enhance Your Calm'
exceptions
NIFI-5953 Manage GetTwitter connection retries on '420 Enhance Your Calm'
exceptions
Update "Max Client Error Retries" parameter name.
reintriduce client.reconnect() on HTTP_ERROR 420
This closes #3276.
Signed-off-by: Koji Kawamura <[email protected]>
---
.../org/apache/nifi/processors/twitter/GetTwitter.java | 14 ++++++++++++++
.../org/apache/nifi/processors/twitter/TestGetTwitter.java | 5 +++++
2 files changed, 19 insertions(+)
diff --git
a/nifi-nar-bundles/nifi-social-media-bundle/nifi-twitter-processors/src/main/java/org/apache/nifi/processors/twitter/GetTwitter.java
b/nifi-nar-bundles/nifi-social-media-bundle/nifi-twitter-processors/src/main/java/org/apache/nifi/processors/twitter/GetTwitter.java
index f0a8b0c..dafa9d0 100644
---
a/nifi-nar-bundles/nifi-social-media-bundle/nifi-twitter-processors/src/main/java/org/apache/nifi/processors/twitter/GetTwitter.java
+++
b/nifi-nar-bundles/nifi-social-media-bundle/nifi-twitter-processors/src/main/java/org/apache/nifi/processors/twitter/GetTwitter.java
@@ -85,6 +85,17 @@ public class GetTwitter extends AbstractProcessor {
.allowableValues(ENDPOINT_SAMPLE, ENDPOINT_FIREHOSE,
ENDPOINT_FILTER)
.defaultValue(ENDPOINT_SAMPLE.getValue())
.build();
+ public static final PropertyDescriptor MAX_CLIENT_ERROR_RETRIES = new
PropertyDescriptor.Builder()
+ .name("max-client-error-retries")
+ .displayName("Max Client Error Retries")
+ .description("The maximum number of retries to attempt when client
experience retryable connection errors."
+ + " Client continues attempting to reconnect using an
exponential back-off pattern until it successfully reconnects"
+ + " or until it reaches the retry limit."
+ +" It is recommended to raise this value when client is
getting rate limited by Twitter API. Default value is 5.")
+ .required(true)
+ .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
+ .defaultValue("5")
+ .build();
public static final PropertyDescriptor CONSUMER_KEY = new
PropertyDescriptor.Builder()
.name("Consumer Key")
.description("The Consumer Key provided by Twitter")
@@ -161,6 +172,7 @@ public class GetTwitter extends AbstractProcessor {
protected void init(final ProcessorInitializationContext context) {
final List<PropertyDescriptor> descriptors = new ArrayList<>();
descriptors.add(ENDPOINT);
+ descriptors.add(MAX_CLIENT_ERROR_RETRIES);
descriptors.add(CONSUMER_KEY);
descriptors.add(CONSUMER_SECRET);
descriptors.add(ACCESS_TOKEN);
@@ -222,6 +234,7 @@ public class GetTwitter extends AbstractProcessor {
@OnScheduled
public void onScheduled(final ProcessContext context) {
final String endpointName = context.getProperty(ENDPOINT).getValue();
+ final int maxRetries =
context.getProperty(MAX_CLIENT_ERROR_RETRIES).asInteger().intValue();
final Authentication oauth = new
OAuth1(context.getProperty(CONSUMER_KEY).getValue(),
context.getProperty(CONSUMER_SECRET).getValue(),
context.getProperty(ACCESS_TOKEN).getValue(),
@@ -319,6 +332,7 @@ public class GetTwitter extends AbstractProcessor {
}
clientBuilder.hosts(host).endpoint(streamingEndpoint);
+ clientBuilder.retries(maxRetries);
client = clientBuilder.build();
client.connect();
}
diff --git
a/nifi-nar-bundles/nifi-social-media-bundle/nifi-twitter-processors/src/test/java/org/apache/nifi/processors/twitter/TestGetTwitter.java
b/nifi-nar-bundles/nifi-social-media-bundle/nifi-twitter-processors/src/test/java/org/apache/nifi/processors/twitter/TestGetTwitter.java
index e56be67..6124f34 100644
---
a/nifi-nar-bundles/nifi-social-media-bundle/nifi-twitter-processors/src/test/java/org/apache/nifi/processors/twitter/TestGetTwitter.java
+++
b/nifi-nar-bundles/nifi-social-media-bundle/nifi-twitter-processors/src/test/java/org/apache/nifi/processors/twitter/TestGetTwitter.java
@@ -26,6 +26,7 @@ public class TestGetTwitter {
public void testLocationValidatorWithValidLocations() {
final TestRunner runner = TestRunners.newTestRunner(GetTwitter.class);
runner.setProperty(GetTwitter.ENDPOINT, GetTwitter.ENDPOINT_FILTER);
+ runner.setProperty(GetTwitter.MAX_CLIENT_ERROR_RETRIES, "5");
runner.setProperty(GetTwitter.CONSUMER_KEY, "consumerKey");
runner.setProperty(GetTwitter.CONSUMER_SECRET, "consumerSecret");
runner.setProperty(GetTwitter.ACCESS_TOKEN, "accessToken");
@@ -38,6 +39,7 @@ public class TestGetTwitter {
public void testLocationValidatorWithEqualLatitudes() {
final TestRunner runner = TestRunners.newTestRunner(GetTwitter.class);
runner.setProperty(GetTwitter.ENDPOINT, GetTwitter.ENDPOINT_FILTER);
+ runner.setProperty(GetTwitter.MAX_CLIENT_ERROR_RETRIES, "5");
runner.setProperty(GetTwitter.CONSUMER_KEY, "consumerKey");
runner.setProperty(GetTwitter.CONSUMER_SECRET, "consumerSecret");
runner.setProperty(GetTwitter.ACCESS_TOKEN, "accessToken");
@@ -50,6 +52,7 @@ public class TestGetTwitter {
public void testLocationValidatorWithEqualLongitudes() {
final TestRunner runner = TestRunners.newTestRunner(GetTwitter.class);
runner.setProperty(GetTwitter.ENDPOINT, GetTwitter.ENDPOINT_FILTER);
+ runner.setProperty(GetTwitter.MAX_CLIENT_ERROR_RETRIES, "5");
runner.setProperty(GetTwitter.CONSUMER_KEY, "consumerKey");
runner.setProperty(GetTwitter.CONSUMER_SECRET, "consumerSecret");
runner.setProperty(GetTwitter.ACCESS_TOKEN, "accessToken");
@@ -62,6 +65,7 @@ public class TestGetTwitter {
public void testLocationValidatorWithSWLatGreaterThanNELat() {
final TestRunner runner = TestRunners.newTestRunner(GetTwitter.class);
runner.setProperty(GetTwitter.ENDPOINT, GetTwitter.ENDPOINT_FILTER);
+ runner.setProperty(GetTwitter.MAX_CLIENT_ERROR_RETRIES, "5");
runner.setProperty(GetTwitter.CONSUMER_KEY, "consumerKey");
runner.setProperty(GetTwitter.CONSUMER_SECRET, "consumerSecret");
runner.setProperty(GetTwitter.ACCESS_TOKEN, "accessToken");
@@ -74,6 +78,7 @@ public class TestGetTwitter {
public void testLocationValidatorWithSWLonGreaterThanNELon() {
final TestRunner runner = TestRunners.newTestRunner(GetTwitter.class);
runner.setProperty(GetTwitter.ENDPOINT, GetTwitter.ENDPOINT_FILTER);
+ runner.setProperty(GetTwitter.MAX_CLIENT_ERROR_RETRIES, "5");
runner.setProperty(GetTwitter.CONSUMER_KEY, "consumerKey");
runner.setProperty(GetTwitter.CONSUMER_SECRET, "consumerSecret");
runner.setProperty(GetTwitter.ACCESS_TOKEN, "accessToken");
