You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@streams.apache.org by robdouglas <gi...@git.apache.org> on 2014/05/19 20:43:19 UTC
[GitHub] incubator-streams pull request: Streams 86
GitHub user robdouglas opened a pull request:
https://github.com/apache/incubator-streams/pull/17
Streams 86
Added in Reentrant read and write locks anytime the provider queue is being updated or read from. This should ensure that we do not have situations where data is lost between this provider and any subsequent components down stream from it.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/robdouglas/incubator-streams STREAMS-86
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/incubator-streams/pull/17.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #17
----
commit 50a70bcd9504dab60a9b694b3f25f738d09ab595
Author: Robert Douglas <rd...@w2odigital.com>
Date: 2014-05-19T18:33:47Z
STREAMS-86 | Added a Reentrant Read/Write lock to ensure that the provider queue is only accessed when appropriate. This helps to make sure that we don't drop data between this provider any any subsequent components
commit 6ccf4e6d8dd83695e65a81dc69bc661ee4840da7
Author: Robert Douglas <rd...@w2odigital.com>
Date: 2014-05-19T18:37:52Z
STREAMS-86 | Added a readlock around offering to the provider queue
commit a08900b6eb753c2bf57c0d7047bda84ca4ebe69f
Author: Robert Douglas <rd...@w2odigital.com>
Date: 2014-05-19T18:40:09Z
STREAMS-86 | Added back in old number of attempts to try (will change in separate pull request)
commit 34bb2bc7c40ac351ff4f0cad0f80584fe58aff5b
Author: Robert Douglas <rd...@w2odigital.com>
Date: 2014-05-19T18:42:03Z
STREAMS-86 | Added in constructQueue() call in prepare method
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: Streams 86
Posted by smashew <gi...@git.apache.org>.
Github user smashew commented on a diff in the pull request:
https://github.com/apache/incubator-streams/pull/17#discussion_r12812291
--- Diff: streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterTimelineProvider.java ---
@@ -45,6 +62,7 @@
private TwitterStreamConfiguration config;
private Class klass;
+ private ReadWriteLock lock = new ReentrantReadWriteLock();
--- End diff --
might as well make this final so you don't accidentally overwrite the lock during execution. Also allows for it to be farmed out to other threads if necessary.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: Streams 86
Posted by robdouglas <gi...@git.apache.org>.
Github user robdouglas commented on a diff in the pull request:
https://github.com/apache/incubator-streams/pull/17#discussion_r12811972
--- Diff: streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterTimelineProvider.java ---
@@ -54,12 +72,14 @@ public void setConfig(TwitterStreamConfiguration config) {
this.config = config;
}
- protected final Queue<StreamsDatum> providerQueue = Queues.synchronizedQueue(new ArrayBlockingQueue<StreamsDatum>(5000));
+ protected volatile Queue<StreamsDatum> providerQueue;
protected int idsCount;
protected Twitter client;
--- End diff --
Would that flexibility essentially be allowing an overridden version of captureTwitterTimeline to use some sort of different queue than the one that is defined here?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: Streams 86
Posted by smashew <gi...@git.apache.org>.
Github user smashew commented on a diff in the pull request:
https://github.com/apache/incubator-streams/pull/17#discussion_r12811784
--- Diff: streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterTimelineProvider.java ---
@@ -54,12 +72,14 @@ public void setConfig(TwitterStreamConfiguration config) {
this.config = config;
}
- protected final Queue<StreamsDatum> providerQueue = Queues.synchronizedQueue(new ArrayBlockingQueue<StreamsDatum>(5000));
+ protected volatile Queue<StreamsDatum> providerQueue;
protected int idsCount;
protected Twitter client;
--- End diff --
This should be a local variable to CaptureTimeline for flexibility.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: Streams 86
Posted by smashew <gi...@git.apache.org>.
Github user smashew commented on a diff in the pull request:
https://github.com/apache/incubator-streams/pull/17#discussion_r12812452
--- Diff: streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterTimelineProvider.java ---
@@ -246,9 +281,15 @@ public void prepare(Object o) {
includeEntitiesEnabled = Optional.fromNullable(new Boolean(Boolean.parseBoolean(config.getIncludeEntities()))).or(true);
client = getTwitterClient();
+
+ try {
+ lock.writeLock().lock();
+ providerQueue = constructQueue();
+ } finally {
+ lock.writeLock().unlock();
--- End diff --
Synchronizing on the class would be cumbersome. Presumes that only 1 stream that uses this class would ever be in use. So, it'd affect threads outside of the one stream being run. Given the circumstances, the local variable lock is appropriate, but should probably be final to ensure it isn't accidentally re-created after this classes instance is created.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: Streams 86
Posted by mfranklin <gi...@git.apache.org>.
Github user mfranklin commented on a diff in the pull request:
https://github.com/apache/incubator-streams/pull/17#discussion_r12810177
--- Diff: streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterTimelineProvider.java ---
@@ -246,9 +281,11 @@ public void prepare(Object o) {
includeEntitiesEnabled = Optional.fromNullable(new Boolean(Boolean.parseBoolean(config.getIncludeEntities()))).or(true);
client = getTwitterClient();
+
+ providerQueue = constructQueue();
--- End diff --
any time you replace the reference to the queue, you should do a write lock.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: Streams 86
Posted by mfranklin <gi...@git.apache.org>.
Github user mfranklin commented on the pull request:
https://github.com/apache/incubator-streams/pull/17#issuecomment-44279875
This is applied. Please close
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: Streams 86
Posted by smashew <gi...@git.apache.org>.
Github user smashew commented on a diff in the pull request:
https://github.com/apache/incubator-streams/pull/17#discussion_r12811736
--- Diff: streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterTimelineProvider.java ---
@@ -116,17 +136,20 @@ protected void captureTimeline(long currentId) {
int keepTrying = 0;
// keep trying to load, give it 5 attempts.
- //while (keepTrying < 10)
while (keepTrying < 1)
--- End diff --
This line should keep trying a few times, to let the back-off do it's work or if Twitter fails.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: Streams 86
Posted by robdouglas <gi...@git.apache.org>.
Github user robdouglas closed the pull request at:
https://github.com/apache/incubator-streams/pull/17
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: Streams 86
Posted by smashew <gi...@git.apache.org>.
Github user smashew commented on a diff in the pull request:
https://github.com/apache/incubator-streams/pull/17#discussion_r12812241
--- Diff: streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterTimelineProvider.java ---
@@ -54,12 +72,14 @@ public void setConfig(TwitterStreamConfiguration config) {
this.config = config;
}
- protected final Queue<StreamsDatum> providerQueue = Queues.synchronizedQueue(new ArrayBlockingQueue<StreamsDatum>(5000));
+ protected volatile Queue<StreamsDatum> providerQueue;
--- End diff --
this doesn't need to be volatile... you have Locks around it being changed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: Streams 86
Posted by mfranklin <gi...@git.apache.org>.
Github user mfranklin commented on a diff in the pull request:
https://github.com/apache/incubator-streams/pull/17#discussion_r12812492
--- Diff: streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterTimelineProvider.java ---
@@ -246,9 +281,15 @@ public void prepare(Object o) {
includeEntitiesEnabled = Optional.fromNullable(new Boolean(Boolean.parseBoolean(config.getIncludeEntities()))).or(true);
client = getTwitterClient();
+
+ try {
+ lock.writeLock().lock();
+ providerQueue = constructQueue();
+ } finally {
+ lock.writeLock().unlock();
--- End diff --
Agreed on finality.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: Streams 86
Posted by mfranklin <gi...@git.apache.org>.
Github user mfranklin commented on a diff in the pull request:
https://github.com/apache/incubator-streams/pull/17#discussion_r12812061
--- Diff: streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterTimelineProvider.java ---
@@ -246,9 +281,15 @@ public void prepare(Object o) {
includeEntitiesEnabled = Optional.fromNullable(new Boolean(Boolean.parseBoolean(config.getIncludeEntities()))).or(true);
client = getTwitterClient();
+
+ try {
+ lock.writeLock().lock();
+ providerQueue = constructQueue();
+ } finally {
+ lock.writeLock().unlock();
--- End diff --
The naming of the classes is the issue, but those are outside of our control. You can synchronize all of the blocks on the same object for better readability in this class, as there will only ever be one thread adding to the queue.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: Streams 86
Posted by smashew <gi...@git.apache.org>.
Github user smashew commented on a diff in the pull request:
https://github.com/apache/incubator-streams/pull/17#discussion_r12811822
--- Diff: streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterTimelineProvider.java ---
@@ -246,9 +281,15 @@ public void prepare(Object o) {
includeEntitiesEnabled = Optional.fromNullable(new Boolean(Boolean.parseBoolean(config.getIncludeEntities()))).or(true);
client = getTwitterClient();
+
+ try {
+ lock.writeLock().lock();
+ providerQueue = constructQueue();
+ } finally {
+ lock.writeLock().unlock();
--- End diff --
While I understand this is necessary under the current paradigm, I don't agree with it.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: Streams 86
Posted by mfranklin <gi...@git.apache.org>.
Github user mfranklin commented on the pull request:
https://github.com/apache/incubator-streams/pull/17#issuecomment-43544336
:+1: I will commit this tomorrow, unless there are objections
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: Streams 86
Posted by smashew <gi...@git.apache.org>.
Github user smashew commented on a diff in the pull request:
https://github.com/apache/incubator-streams/pull/17#discussion_r12812370
--- Diff: streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterTimelineProvider.java ---
@@ -54,12 +72,14 @@ public void setConfig(TwitterStreamConfiguration config) {
this.config = config;
}
- protected final Queue<StreamsDatum> providerQueue = Queues.synchronizedQueue(new ArrayBlockingQueue<StreamsDatum>(5000));
+ protected volatile Queue<StreamsDatum> providerQueue;
protected int idsCount;
protected Twitter client;
--- End diff --
in your method below, it is being treated as an instance variable. with only one function really needing to access it. Refactoring out the private just keeps memory issues to a min.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-streams pull request: Streams 86
Posted by smashew <gi...@git.apache.org>.
Github user smashew commented on a diff in the pull request:
https://github.com/apache/incubator-streams/pull/17#discussion_r12812532
--- Diff: streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterTimelineProvider.java ---
@@ -246,9 +281,15 @@ public void prepare(Object o) {
includeEntitiesEnabled = Optional.fromNullable(new Boolean(Boolean.parseBoolean(config.getIncludeEntities()))).or(true);
client = getTwitterClient();
+
+ try {
+ lock.writeLock().lock();
+ providerQueue = constructQueue();
+ } finally {
+ lock.writeLock().unlock();
--- End diff --
+1 There is a similar issue in the ElasticSearchReader
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---