You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2022/05/09 08:32:06 UTC
[GitHub] [kafka] cadonna opened a new pull request, #12137: DRAFT: Consolidate StreamsException and TaskCorruptedException
cadonna opened a new pull request, #12137:
URL: https://github.com/apache/kafka/pull/12137
The TaskCorruptedException extends StreamsException and both exceptions
have the capability to store the task IDs that caused the exception. The
only difference is that StreamsException only stores one single exception
whereas TaskCorruptedException stores multiple exception.
This PR proposed to consolidate the two exception by storing multiple
task IDs in StreamsException.
DO NOT MERGE THIS PR, since it needs a KIP.
If folks agree with my proposal I will create a KIP.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna commented on pull request #12137: MINOR: Consolidate StreamsException and TaskCorruptedException
Posted by GitBox <gi...@apache.org>.
cadonna commented on PR #12137:
URL: https://github.com/apache/kafka/pull/12137#issuecomment-1128674513
I will think about it! Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna commented on a diff in pull request #12137: MINOR: Consolidate StreamsException and TaskCorruptedException
Posted by GitBox <gi...@apache.org>.
cadonna commented on code in PR #12137:
URL: https://github.com/apache/kafka/pull/12137#discussion_r870089994
##########
streams/src/main/java/org/apache/kafka/streams/errors/StreamsException.java:
##########
@@ -57,19 +73,29 @@ public StreamsException(final Throwable throwable) {
public StreamsException(final Throwable throwable, final TaskId taskId) {
super(throwable);
- this.taskId = taskId;
+ this.taskIds.add(taskId);
}
/**
* @return the {@link TaskId} that this exception originated from, or {@link Optional#empty()} if the exception
* cannot be traced back to a particular task. Note that the {@code TaskId} being empty does not
* guarantee that the exception wasn't directly related to a specific task.
*/
+ @Deprecated
public Optional<TaskId> taskId() {
- return Optional.ofNullable(taskId);
+ return taskIds.isEmpty() ? Optional.empty() : Optional.of(new ArrayList<>(taskIds).get(0));
+ }
+
+ public Set<TaskId> taskIds() {
Review Comment:
See above.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mjsax commented on a diff in pull request #12137: DRAFT: Consolidate StreamsException and TaskCorruptedException
Posted by GitBox <gi...@apache.org>.
mjsax commented on code in PR #12137:
URL: https://github.com/apache/kafka/pull/12137#discussion_r868282499
##########
streams/src/main/java/org/apache/kafka/streams/errors/StreamsException.java:
##########
@@ -31,24 +37,34 @@ public class StreamsException extends KafkaException {
private final static long serialVersionUID = 1L;
- private TaskId taskId;
+ private final Set<TaskId> taskIds = new HashSet<>();
public StreamsException(final String message) {
this(message, (TaskId) null);
}
public StreamsException(final String message, final TaskId taskId) {
super(message);
- this.taskId = taskId;
+ taskIds.add(taskId);
+ }
+
+ public StreamsException(final String message, final Set<TaskId> taskIds) {
Review Comment:
Should this be `protected` or package-private instead of `public` ?
##########
streams/src/main/java/org/apache/kafka/streams/errors/StreamsException.java:
##########
@@ -57,19 +73,29 @@ public StreamsException(final Throwable throwable) {
public StreamsException(final Throwable throwable, final TaskId taskId) {
super(throwable);
- this.taskId = taskId;
+ this.taskIds.add(taskId);
}
/**
* @return the {@link TaskId} that this exception originated from, or {@link Optional#empty()} if the exception
* cannot be traced back to a particular task. Note that the {@code TaskId} being empty does not
* guarantee that the exception wasn't directly related to a specific task.
*/
+ @Deprecated
Review Comment:
Why do we need to deprecate this method? (Same question below).
##########
streams/src/main/java/org/apache/kafka/streams/errors/StreamsException.java:
##########
@@ -57,19 +73,29 @@ public StreamsException(final Throwable throwable) {
public StreamsException(final Throwable throwable, final TaskId taskId) {
super(throwable);
- this.taskId = taskId;
+ this.taskIds.add(taskId);
}
/**
* @return the {@link TaskId} that this exception originated from, or {@link Optional#empty()} if the exception
* cannot be traced back to a particular task. Note that the {@code TaskId} being empty does not
* guarantee that the exception wasn't directly related to a specific task.
*/
+ @Deprecated
public Optional<TaskId> taskId() {
- return Optional.ofNullable(taskId);
+ return taskIds.isEmpty() ? Optional.empty() : Optional.of(new ArrayList<>(taskIds).get(0));
+ }
+
+ public Set<TaskId> taskIds() {
Review Comment:
Why do we need to add this method? (Same question below).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna closed pull request #12137: MINOR: Consolidate StreamsException and TaskCorruptedException
Posted by GitBox <gi...@apache.org>.
cadonna closed pull request #12137: MINOR: Consolidate StreamsException and TaskCorruptedException
URL: https://github.com/apache/kafka/pull/12137
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mjsax commented on pull request #12137: MINOR: Consolidate StreamsException and TaskCorruptedException
Posted by GitBox <gi...@apache.org>.
mjsax commented on PR #12137:
URL: https://github.com/apache/kafka/pull/12137#issuecomment-1128295416
> Avoiding the public API change seems quite hacky to me
I leave it up to you -- it's only a small KIP so if you want to do it, just go for it -- personally, I don't see much benefit in unifying both exceptions (also indicated that this PR doesn't change any code that uses them). But it also does not hurt...
Your call.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna commented on a diff in pull request #12137: MINOR: Consolidate StreamsException and TaskCorruptedException
Posted by GitBox <gi...@apache.org>.
cadonna commented on code in PR #12137:
URL: https://github.com/apache/kafka/pull/12137#discussion_r870087908
##########
streams/src/main/java/org/apache/kafka/streams/errors/StreamsException.java:
##########
@@ -31,24 +37,34 @@ public class StreamsException extends KafkaException {
private final static long serialVersionUID = 1L;
- private TaskId taskId;
+ private final Set<TaskId> taskIds = new HashSet<>();
public StreamsException(final String message) {
this(message, (TaskId) null);
}
public StreamsException(final String message, final TaskId taskId) {
super(message);
- this.taskId = taskId;
+ taskIds.add(taskId);
+ }
+
+ public StreamsException(final String message, final Set<TaskId> taskIds) {
Review Comment:
Switched to package-private.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna commented on pull request #12137: MINOR: Consolidate StreamsException and TaskCorruptedException
Posted by GitBox <gi...@apache.org>.
cadonna commented on PR #12137:
URL: https://github.com/apache/kafka/pull/12137#issuecomment-1123449313
Avoiding the public API change seems quite hacky to me. I would prefer to generalize `StreamsException` to hold multiple task IDs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna commented on a diff in pull request #12137: MINOR: Consolidate StreamsException and TaskCorruptedException
Posted by GitBox <gi...@apache.org>.
cadonna commented on code in PR #12137:
URL: https://github.com/apache/kafka/pull/12137#discussion_r870087244
##########
streams/src/main/java/org/apache/kafka/streams/errors/StreamsException.java:
##########
@@ -66,10 +80,15 @@ public StreamsException(final Throwable throwable, final TaskId taskId) {
* guarantee that the exception wasn't directly related to a specific task.
*/
public Optional<TaskId> taskId() {
- return Optional.ofNullable(taskId);
+ return taskIds.isEmpty() ? Optional.empty() : Optional.ofNullable(new ArrayList<>(taskIds).get(0));
}
public void setTaskId(final TaskId taskId) {
- this.taskId = taskId;
+ taskIds.clear();
Review Comment:
I need to clear the tasks IDs otherwise you might not be able to get the task ID with `taskId()` after the first call to `setTaskId()`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cadonna commented on a diff in pull request #12137: MINOR: Consolidate StreamsException and TaskCorruptedException
Posted by GitBox <gi...@apache.org>.
cadonna commented on code in PR #12137:
URL: https://github.com/apache/kafka/pull/12137#discussion_r870089590
##########
streams/src/main/java/org/apache/kafka/streams/errors/StreamsException.java:
##########
@@ -57,19 +73,29 @@ public StreamsException(final Throwable throwable) {
public StreamsException(final Throwable throwable, final TaskId taskId) {
super(throwable);
- this.taskId = taskId;
+ this.taskIds.add(taskId);
}
/**
* @return the {@link TaskId} that this exception originated from, or {@link Optional#empty()} if the exception
* cannot be traced back to a particular task. Note that the {@code TaskId} being empty does not
* guarantee that the exception wasn't directly related to a specific task.
*/
+ @Deprecated
Review Comment:
I thought it would be better to generalise `StreamsException` and allow it to hold multiple taskIDs. In that case you would not need a getter for a single task ID.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org