You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@storm.apache.org by srdo <gi...@git.apache.org> on 2016/12/14 18:33:32 UTC
[GitHub] storm pull request #1826: STORM-2087 1.x
GitHub user srdo opened a pull request:
https://github.com/apache/storm/pull/1826
STORM-2087 1.x
Backport STORM-2087 to 1.x
See https://github.com/apache/storm/pull/1679 for master version
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/srdo/storm STORM-2087-1.x
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/storm/pull/1826.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 #1826
----
commit 5c17a59ebf9f5667ccfb2dc0a127044b94a0de86
Author: Jeff Fenchel <jf...@gmail.com>
Date: 2016-09-11T01:32:35Z
STORM-2087: storm-kafka-client improvements for kafka 0.10
- Added unit tests to storm-kafka-client
- Fixed bug in kafka-spout-client that resulted in tuples not being replayed
when a failure occurred immediately after the last commit
- Modified the kafka spout to continue processing and not halt upon receiving
a double ack
commit cc39c679dd00a27c4f42baf340be0acffec8c8d0
Author: Stig Rohde D�ssing <sd...@it-minds.dk>
Date: 2016-12-14T18:29:48Z
STORM-2087: Make tests Java 7 compatible
----
---
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] storm pull request #1826: STORM-2087 1.x
Posted by hmcl <gi...@git.apache.org>.
Github user hmcl commented on a diff in the pull request:
https://github.com/apache/storm/pull/1826#discussion_r92919073
--- Diff: external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java ---
@@ -77,7 +77,7 @@
private KafkaSpoutStreams kafkaSpoutStreams; // Object that wraps all the logic to declare output fields and emit tuples
private transient KafkaSpoutTuplesBuilder<K, V> tuplesBuilder; // Object that contains the logic to build tuples for each ConsumerRecord
- private transient Map<TopicPartition, OffsetEntry> acked; // Tuples that were successfully acked. These tuples will be committed periodically when the commit timer expires, after consumer rebalance, or on close/deactivate
+ transient Map<TopicPartition, OffsetEntry> acked; // Tuples that were successfully acked. These tuples will be committed periodically when the commit timer expires, after consumer rebalance, or on close/deactivate
--- End diff --
@srdo using reflection for tests only wouldn't look to concerning to me, but I also agree with your point of not being able to do find usages in the IDE. I would also say that if the tests and the source class are correctly implemented, the tests should never need to query the internal state to assert of the correctness of the code. If one has to do that, that to me feels like something is not as good as it could be.
In my opinion package private methods have a very specific use, which is for the construction of libraries. Although they are not part of the public API, they are easy to misuse by clients who may want to take shortcuts. With this said, I think that having a package where we put some of the nested classes would make the code significantly more modular. I really think that it is getting to the point where I start seeing `if / else` blocks all over the place, added to `instance of` ... and I really want to avoid that at all costs... otherwise soon we wont' have a hand on this.
However extracting the nested classes will also add an extra complexity, because there is a significant amount of state that the `OffsetEntry` class uses that would have to be passed around if it becomes it's own class. I still thing we should do it, though. When we do it should be in such a way that we can plugin a different implementation of `OffsetEntry`. Perhaps also rename this class, as it is really an `OffsetManager`. What do you think ?
---
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] storm issue #1826: STORM-2087 1.x
Posted by jfenc91 <gi...@git.apache.org>.
Github user jfenc91 commented on the issue:
https://github.com/apache/storm/pull/1826
Thanks for picking this up @srdo!
---
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] storm pull request #1826: STORM-2087 1.x
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/storm/pull/1826
---
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] storm pull request #1826: STORM-2087 1.x
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/1826#discussion_r92928193
--- Diff: external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java ---
@@ -77,7 +77,7 @@
private KafkaSpoutStreams kafkaSpoutStreams; // Object that wraps all the logic to declare output fields and emit tuples
private transient KafkaSpoutTuplesBuilder<K, V> tuplesBuilder; // Object that contains the logic to build tuples for each ConsumerRecord
- private transient Map<TopicPartition, OffsetEntry> acked; // Tuples that were successfully acked. These tuples will be committed periodically when the commit timer expires, after consumer rebalance, or on close/deactivate
+ transient Map<TopicPartition, OffsetEntry> acked; // Tuples that were successfully acked. These tuples will be committed periodically when the commit timer expires, after consumer rebalance, or on close/deactivate
--- End diff --
@hmcl The changes are here https://github.com/apache/storm/pull/1832. I'd like to merge this branch, then we can backport the refactoring in a separate PR once it has been merged to master.
---
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] storm pull request #1826: STORM-2087 1.x
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/1826#discussion_r92676777
--- Diff: external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java ---
@@ -77,7 +77,7 @@
private KafkaSpoutStreams kafkaSpoutStreams; // Object that wraps all the logic to declare output fields and emit tuples
private transient KafkaSpoutTuplesBuilder<K, V> tuplesBuilder; // Object that contains the logic to build tuples for each ConsumerRecord
- private transient Map<TopicPartition, OffsetEntry> acked; // Tuples that were successfully acked. These tuples will be committed periodically when the commit timer expires, after consumer rebalance, or on close/deactivate
+ transient Map<TopicPartition, OffsetEntry> acked; // Tuples that were successfully acked. These tuples will be committed periodically when the commit timer expires, after consumer rebalance, or on close/deactivate
--- End diff --
@hmcl (sorry, forgot to ping you on the other comment) The same applies to Timer btw, thinking of your comment about the spout code getting a little big. We could easily move that to the internal package too.
---
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] storm pull request #1826: STORM-2087 1.x
Posted by hmcl <gi...@git.apache.org>.
Github user hmcl commented on a diff in the pull request:
https://github.com/apache/storm/pull/1826#discussion_r92546386
--- Diff: external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java ---
@@ -77,7 +77,7 @@
private KafkaSpoutStreams kafkaSpoutStreams; // Object that wraps all the logic to declare output fields and emit tuples
private transient KafkaSpoutTuplesBuilder<K, V> tuplesBuilder; // Object that contains the logic to build tuples for each ConsumerRecord
- private transient Map<TopicPartition, OffsetEntry> acked; // Tuples that were successfully acked. These tuples will be committed periodically when the commit timer expires, after consumer rebalance, or on close/deactivate
+ transient Map<TopicPartition, OffsetEntry> acked; // Tuples that were successfully acked. These tuples will be committed periodically when the commit timer expires, after consumer rebalance, or on close/deactivate
--- End diff --
I believe this is for testing purposes, right? There are ways for testing frameworks to access private fields, if really necessary. I am in favor of not break encapsulation and expose internal state for the purposes of testing.
---
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] storm pull request #1826: STORM-2087 1.x
Posted by srdo <gi...@git.apache.org>.
Github user srdo commented on a diff in the pull request:
https://github.com/apache/storm/pull/1826#discussion_r92921273
--- Diff: external/storm-kafka-client/src/main/java/org/apache/storm/kafka/spout/KafkaSpout.java ---
@@ -77,7 +77,7 @@
private KafkaSpoutStreams kafkaSpoutStreams; // Object that wraps all the logic to declare output fields and emit tuples
private transient KafkaSpoutTuplesBuilder<K, V> tuplesBuilder; // Object that contains the logic to build tuples for each ConsumerRecord
- private transient Map<TopicPartition, OffsetEntry> acked; // Tuples that were successfully acked. These tuples will be committed periodically when the commit timer expires, after consumer rebalance, or on close/deactivate
+ transient Map<TopicPartition, OffsetEntry> acked; // Tuples that were successfully acked. These tuples will be committed periodically when the commit timer expires, after consumer rebalance, or on close/deactivate
--- End diff --
Yes, I agree that we should move OffsetEntry out of KafkaSpout. I also agree that we should be able to test the spout without peeking directly into the spout's fields.
Here's what I suggest:
* We move Timer out into a separate class, and use a TimerFactory to create instances in the KafkaSpout (basically the same solution as we use for KafkaConsumer mocking on master). That way we can write tests where time is simulated, which is helpful if we want to test that the spout commits offsets correctly (or any other periodic action).
* We move and rename OffsetEntry out into a new class. It looks like the only part of KafkaSpout state it reads is `numUncommittedOffsets` from `OffsetEntry.commit()`. Since that method returns void, we can solve that problem very easily by returning `numCommittedOffsets` from there. If we want to support other OffsetEntry implementations we can just set up a factory for that too, but I think that can be postponed until we actually need to support multiple implementations.
* We rewrite the offending tests here to use a mocked KafkaConsumer and KafkaTimer instead of KafkaUnit, and we make the tests check that the offsets are committed on the consumer instead of directly reading `KafkaSpout.acked`. I think KafkaUnit is a good idea for integration tests, but I don't think it fits that well for the tests in this PR.
I'll try implementing this on master, if it works out I'll open a PR there first. We can port the improvements to 1.x once we agree on the solution on master.
---
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] storm issue #1826: STORM-2087 1.x
Posted by HeartSaVioR <gi...@git.apache.org>.
Github user HeartSaVioR commented on the issue:
https://github.com/apache/storm/pull/1826
+1 since follow-up issue and PR are created so good to merge as it is.
---
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.
---