STORM-2087 1.x #1826
STORM-2087 1.x #1826
Conversation
- 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
| @@ -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 | |||
There was a problem hiding this comment.
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.
There was a problem hiding this comment.
I agree to some extent, though this also needs to be fixed on master if we change it here. Breaking encapsulation with package private fields is not a big issue IMO since it doesn't change the public API. Also it probably wouldn't hurt to split some of the code into smaller packages, which limits visibility.
I'd really like to avoid using reflection if possible. It's brittle and prevents the IDE from helping you when you're refactoring (renaming a field for example).
How would you feel about adding acked to the testing constructor and injecting it from the tests instead? That way the tests still have access to it without the field being package private. We could also pretty easily move OffsetEntry out into the spout.internal package.
There was a problem hiding this comment.
@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.
There was a problem hiding this comment.
@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 ?
There was a problem hiding this comment.
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
numUncommittedOffsetsfromOffsetEntry.commit(). Since that method returns void, we can solve that problem very easily by returningnumCommittedOffsetsfrom 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.
|
Thanks for picking this up @srdo! |
|
+1 since follow-up issue and PR are created so good to merge as it is. |
Backport STORM-2087 to 1.x
See #1679 for master version