You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@storm.apache.org by Parth-Brahmbhatt <gi...@git.apache.org> on 2014/12/06 00:14:21 UTC
[GitHub] storm pull request: STORM-586: TridentKafkaEmitter should catch up...
GitHub user Parth-Brahmbhatt opened a pull request:
https://github.com/apache/storm/pull/339
STORM-586: TridentKafkaEmitter should catch updateOffsetException.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/Parth-Brahmbhatt/incubator-storm STORM-586
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/storm/pull/339.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 #339
----
commit 65e9f0c814b2cddc772880042259b66194fd6fb7
Author: Parth Brahmbhatt <br...@gmail.com>
Date: 2014-12-05T22:48:34Z
STORM-586: TridentKafkaEmitter should catch updateOffsetException.
----
---
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: STORM-586: TridentKafkaEmitter should catch up...
Posted by harshach <gi...@git.apache.org>.
Github user harshach commented on a diff in the pull request:
https://github.com/apache/storm/pull/339#discussion_r21414360
--- Diff: external/storm-kafka/src/jvm/storm/kafka/UpdateOffsetException.java ---
@@ -17,6 +17,9 @@
*/
package storm.kafka;
-public class UpdateOffsetException extends RuntimeException {
--- End diff --
I think UpdateOffsetException shouldn't extend FailedFetchException . They meant to be different exceptions . The problem here is if a method throws UpdateOffsetException and there is a catch for FailedFetchException its get caught in that catch but we want to have different behavior for FailedFetchException and UpdateOffsetException. Also UpdateOffsetException can be named differently, may be TopicOffsetOutOfRangeException and for any other errors use FailedFetchException.
---
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: STORM-586: TridentKafkaEmitter should catch up...
Posted by nathanmarz <gi...@git.apache.org>.
Github user nathanmarz commented on the pull request:
https://github.com/apache/storm/pull/339#issuecomment-67581379
-1
Why are offset out of range exceptions happening when re-emitting a batch? That shouldn't be possible so I'd like to know why this is happening.
Emitting a different batch than the first time actually breaks the contract of transactional spouts – which is unacceptable (though it's fine for opaque transactional spouts). Because it breaks the contract of transactional spouts, it would be better to just error repeatedly than silently do the wrong thing.
The part of this code that catches the error and emits an empty batch for the same batch id should only be applied towards the opaque spout. Additionally, by better understanding why those offset out of range exceptions are happening, there may be an alternative solution that doesn't involve breaking the contract of transactional spouts.
---
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: STORM-586: TridentKafkaEmitter should catch up...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/storm/pull/339
---
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: STORM-586: TridentKafkaEmitter should catch up...
Posted by nathanmarz <gi...@git.apache.org>.
Github user nathanmarz commented on the pull request:
https://github.com/apache/storm/pull/339#issuecomment-67581552
Once this code is no longer applied to the transactional spout, I'll be +1. I'd still like to know why that exception is happening though.
---
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: STORM-586: TridentKafkaEmitter should catch up...
Posted by Parth-Brahmbhatt <gi...@git.apache.org>.
Github user Parth-Brahmbhatt commented on the pull request:
https://github.com/apache/storm/pull/339#issuecomment-67555520
I had a chat with @harshach and he pointed out that we will hit https://github.com/apache/storm/pull/321 if we just use KafkaUtil.getOffset method. I have modified the code so as part of handling the OffsetOutOfRange exception we use the earliestTime instead of using the latestTime when forceFromStart = false which is the default value and can result in data loss.
---
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: STORM-586: TridentKafkaEmitter should catch up...
Posted by Parth-Brahmbhatt <gi...@git.apache.org>.
Github user Parth-Brahmbhatt commented on the pull request:
https://github.com/apache/storm/pull/339#issuecomment-66206449
I also tested this locally and it works fine for OpaqueTridentSpout. The TransactionalSpout tries to reemit a batch and when the offset is out of range there is no way to actually do this. I have defaulted the behavior to empty batch so topology can make progress instead of getting stuck. Let me know if you think it is better to let the user control this and I can add a config value.
---
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: STORM-586: TridentKafkaEmitter should catch up...
Posted by nathanmarz <gi...@git.apache.org>.
Github user nathanmarz commented on the pull request:
https://github.com/apache/storm/pull/339#issuecomment-67589509
Yea, most of this can be addressed by allocating more resources so Kafka has enough retention to not truncate the data before the topology can be fixed.
+1
---
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: STORM-586: TridentKafkaEmitter should catch up...
Posted by harshach <gi...@git.apache.org>.
Github user harshach commented on the pull request:
https://github.com/apache/storm/pull/339#issuecomment-67593637
@nathanmarz there is another case where your topology is running and processed all the existing data and there is no new data came into kafka topic which could cause offset out of range.
---
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: STORM-586: TridentKafkaEmitter should catch up...
Posted by harshach <gi...@git.apache.org>.
Github user harshach commented on a diff in the pull request:
https://github.com/apache/storm/pull/339#discussion_r21874150
--- Diff: external/storm-kafka/src/jvm/storm/kafka/KafkaUtils.java ---
@@ -155,7 +155,7 @@ public void refreshPartitions(Set<Partition> partitions) {
}
}
- public static ByteBufferMessageSet fetchMessages(KafkaConfig config, SimpleConsumer consumer, Partition partition, long offset) throws UpdateOffsetException {
+ public static ByteBufferMessageSet fetchMessages(KafkaConfig config, SimpleConsumer consumer, Partition partition, long offset) throws TopicOffsetOutOfRangeException {
--- End diff --
Sorry I should've mention this earlier but can you add FailedFetchException, RuntimeException to the throws class there. fetchMessages throws above two apart from TopicOffsetOutOfRangeException but we only declare one of the exception.
---
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: STORM-586: TridentKafkaEmitter should catch up...
Posted by Parth-Brahmbhatt <gi...@git.apache.org>.
Github user Parth-Brahmbhatt commented on the pull request:
https://github.com/apache/storm/pull/339#issuecomment-67436156
@harshach added exceptions to throws declaration and upmerged.
---
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: STORM-586: TridentKafkaEmitter should catch up...
Posted by harshach <gi...@git.apache.org>.
Github user harshach commented on the pull request:
https://github.com/apache/storm/pull/339#issuecomment-67672726
@nathanmarz Yes previously it used keep retrying but this is causing too many logs being generated by kafka due to the failed requests. This is fixed in https://issues.apache.org/jira/browse/STORM-511. But STORM-511 patch failed to address Trident part of the code. This PR solves this issue.
---
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: STORM-586: TridentKafkaEmitter should catch up...
Posted by harshach <gi...@git.apache.org>.
Github user harshach commented on the pull request:
https://github.com/apache/storm/pull/339#issuecomment-67557355
Thanks for the quick fix @Parth-Brahmbhatt . +1
---
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: STORM-586: TridentKafkaEmitter should catch up...
Posted by harshach <gi...@git.apache.org>.
Github user harshach commented on the pull request:
https://github.com/apache/storm/pull/339#issuecomment-67527319
Thanks @Parth-Brahmbhatt overall it looks good to me. +1. I'll give a day or two before I merge this in.
---
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: STORM-586: TridentKafkaEmitter should catch up...
Posted by nathanmarz <gi...@git.apache.org>.
Github user nathanmarz commented on the pull request:
https://github.com/apache/storm/pull/339#issuecomment-67615349
Can you elaborate? It should just keep trying that last offset repeatedly. This is a case I tested repeatedly when I originally wrote this, so if this is the case it's a regression.
---
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: STORM-586: TridentKafkaEmitter should catch up...
Posted by Parth-Brahmbhatt <gi...@git.apache.org>.
Github user Parth-Brahmbhatt closed the pull request at:
https://github.com/apache/storm/pull/339
---
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: STORM-586: TridentKafkaEmitter should catch up...
Posted by Parth-Brahmbhatt <gi...@git.apache.org>.
Github user Parth-Brahmbhatt commented on the pull request:
https://github.com/apache/storm/pull/339#issuecomment-67582859
@nathanmarz the re-emit can hit this issue in many pathological cases , bad kafka config, transient network failure causing storm batches to fail repeatedly while kafka queue with low retention getting truncated, very slow topology with really fast kafka producers with low retention rate so mostly around kafka config with lower retention rates.
I have made the change you suggested during reemit we don't catch exception but let the spout fail.
---
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: STORM-586: TridentKafkaEmitter should catch up...
Posted by Parth-Brahmbhatt <gi...@git.apache.org>.
GitHub user Parth-Brahmbhatt reopened a pull request:
https://github.com/apache/storm/pull/339
STORM-586: TridentKafkaEmitter should catch updateOffsetException.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/Parth-Brahmbhatt/incubator-storm STORM-586
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/storm/pull/339.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 #339
----
commit 65e9f0c814b2cddc772880042259b66194fd6fb7
Author: Parth Brahmbhatt <br...@gmail.com>
Date: 2014-12-05T22:48:34Z
STORM-586: TridentKafkaEmitter should catch updateOffsetException.
commit 86839dc6b789045a13cf28cba008e52c4d835fa4
Author: Parth Brahmbhatt <br...@gmail.com>
Date: 2014-12-08T22:49:29Z
Ading special case for retry batch, in case of trident a transaction retry should not jump the offset requested as part of retry.
commit b2f48b41f19398498c7ae41c2059f3685b87ac22
Author: Parth Brahmbhatt <br...@gmail.com>
Date: 2014-12-08T23:06:26Z
Renaming UpdateOffsetException to TopicOffsetOutOfRangeException.
commit fcf31350b62ca0efeeea96c8e1b0134edb81c1eb
Author: Parth Brahmbhatt <br...@gmail.com>
Date: 2014-12-08T23:10:13Z
Reverting back to TopicOffsetOutOfRangeException extends RunTimeException.
----
---
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: STORM-586: TridentKafkaEmitter should catch up...
Posted by ptgoetz <gi...@git.apache.org>.
Github user ptgoetz commented on the pull request:
https://github.com/apache/storm/pull/339#issuecomment-66829121
+1
---
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: STORM-586: TridentKafkaEmitter should catch up...
Posted by Parth-Brahmbhatt <gi...@git.apache.org>.
Github user Parth-Brahmbhatt commented on a diff in the pull request:
https://github.com/apache/storm/pull/339#discussion_r21495048
--- Diff: external/storm-kafka/src/jvm/storm/kafka/UpdateOffsetException.java ---
@@ -17,6 +17,9 @@
*/
package storm.kafka;
-public class UpdateOffsetException extends RuntimeException {
--- End diff --
Renaming done. I feel TopicOffsetOutOfRangeException is just another cause for a fetchFailedException so I created a "is a" relationship but on second thoughts you are right. The caller will have a better experience if it had a retriable and non retryable exception that it does not accidently handle in the same catch block. Reverted back to using RunTimeException.
---
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.
---