kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From guozh...@apache.org
Subject kafka git commit: MINOR: cleanup Kafka Streams exception classes
Date Fri, 06 Jan 2017 19:48:14 GMT
Repository: kafka
Updated Branches:
  refs/heads/trunk e0de3a421 -> 10b330cf0


MINOR: cleanup Kafka Streams exception classes

Author: Matthias J. Sax <matthias@confluent.io>

Reviewers: Damian Guy, Guozhang Wang

Closes #2314 from mjsax/javaDocImprovements


Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/10b330cf
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/10b330cf
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/10b330cf

Branch: refs/heads/trunk
Commit: 10b330cf057a90038407a89fdb59d096da01bc55
Parents: e0de3a4
Author: Matthias J. Sax <matthias@confluent.io>
Authored: Fri Jan 6 11:48:11 2017 -0800
Committer: Guozhang Wang <wangguoz@gmail.com>
Committed: Fri Jan 6 11:48:11 2017 -0800

----------------------------------------------------------------------
 .../errors/InvalidStateStoreException.java      | 21 ++++++++++++++++----
 .../kafka/streams/errors/LockException.java     | 17 ++++++++++------
 .../streams/errors/ProcessorStateException.java | 17 ++++++++++------
 .../kafka/streams/errors/StreamsException.java  | 14 +++++++------
 .../streams/errors/TaskAssignmentException.java | 14 +++++++------
 .../streams/errors/TaskIdFormatException.java   | 20 ++++++++++++++++---
 .../errors/TopologyBuilderException.java        | 16 +++++++++++++--
 7 files changed, 86 insertions(+), 33 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/10b330cf/streams/src/main/java/org/apache/kafka/streams/errors/InvalidStateStoreException.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/errors/InvalidStateStoreException.java
b/streams/src/main/java/org/apache/kafka/streams/errors/InvalidStateStoreException.java
index 37ad580..ca57be0 100644
--- a/streams/src/main/java/org/apache/kafka/streams/errors/InvalidStateStoreException.java
+++ b/streams/src/main/java/org/apache/kafka/streams/errors/InvalidStateStoreException.java
@@ -14,18 +14,31 @@
  */
 package org.apache.kafka.streams.errors;
 
+import org.apache.kafka.common.annotation.InterfaceStability;
+
 /**
- * Indicates that there was a problem when trying to access
- * a {@link org.apache.kafka.streams.processor.StateStore}, i.e, the Store is no longer valid
because it is closed
- * or doesn't exist any more due to a rebalance.
- *
+ * Indicates that there was a problem when trying to access a
+ * {@link org.apache.kafka.streams.processor.StateStore StateStore}, i.e, the Store is no
longer valid because it is
+ * closed or doesn't exist any more due to a rebalance.
+ * <p>
  * These exceptions may be transient, i.e., during a rebalance it won't be possible to query
the stores as they are
  * being (re)-initialized. Once the rebalance has completed the stores will be available
again. Hence, it is valid
  * to backoff and retry when handling this exception.
  */
+@InterfaceStability.Unstable
 public class InvalidStateStoreException extends StreamsException {
 
+    private final static long serialVersionUID = 1L;
+
     public InvalidStateStoreException(final String message) {
         super(message);
     }
+
+    public InvalidStateStoreException(final String message, final Throwable throwable) {
+        super(message, throwable);
+    }
+
+    public InvalidStateStoreException(final Throwable throwable) {
+        super(throwable);
+    }
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/10b330cf/streams/src/main/java/org/apache/kafka/streams/errors/LockException.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/errors/LockException.java b/streams/src/main/java/org/apache/kafka/streams/errors/LockException.java
index 00c75ec..c85fd84 100644
--- a/streams/src/main/java/org/apache/kafka/streams/errors/LockException.java
+++ b/streams/src/main/java/org/apache/kafka/streams/errors/LockException.java
@@ -14,22 +14,27 @@
  */
 package org.apache.kafka.streams.errors;
 
+import org.apache.kafka.common.annotation.InterfaceStability;
+
 /**
  * Indicates that the state store directory lock could not be acquired because another thread
holds the lock.
+ *
+ * @see org.apache.kafka.streams.processor.StateStore
  */
+@InterfaceStability.Unstable
 public class LockException extends StreamsException {
 
-    private static final long serialVersionUID = 1L;
+    private final static long serialVersionUID = 1L;
 
-    public LockException(final String s) {
-        super(s);
+    public LockException(final String message) {
+        super(message);
     }
 
-    public LockException(String s, Throwable throwable) {
-        super(s, throwable);
+    public LockException(final String message, final Throwable throwable) {
+        super(message, throwable);
     }
 
-    public LockException(Throwable throwable) {
+    public LockException(final Throwable throwable) {
         super(throwable);
     }
 

http://git-wip-us.apache.org/repos/asf/kafka/blob/10b330cf/streams/src/main/java/org/apache/kafka/streams/errors/ProcessorStateException.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/errors/ProcessorStateException.java
b/streams/src/main/java/org/apache/kafka/streams/errors/ProcessorStateException.java
index e6f872a..061b605 100644
--- a/streams/src/main/java/org/apache/kafka/streams/errors/ProcessorStateException.java
+++ b/streams/src/main/java/org/apache/kafka/streams/errors/ProcessorStateException.java
@@ -17,22 +17,27 @@
 
 package org.apache.kafka.streams.errors;
 
+import org.apache.kafka.common.annotation.InterfaceStability;
+
 /**
  * Indicates a processor state operation (e.g. put, get) has failed.
+ *
+ * @see org.apache.kafka.streams.processor.StateStore
  */
+@InterfaceStability.Unstable
 public class ProcessorStateException extends StreamsException {
 
-    private static final long serialVersionUID = 1L;
+    private final static long serialVersionUID = 1L;
 
-    public ProcessorStateException(String s) {
-        super(s);
+    public ProcessorStateException(final String message) {
+        super(message);
     }
 
-    public ProcessorStateException(String s, Throwable throwable) {
-        super(s, throwable);
+    public ProcessorStateException(final String message, final Throwable throwable) {
+        super(message, throwable);
     }
 
-    public ProcessorStateException(Throwable throwable) {
+    public ProcessorStateException(final Throwable throwable) {
         super(throwable);
     }
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/10b330cf/streams/src/main/java/org/apache/kafka/streams/errors/StreamsException.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/errors/StreamsException.java b/streams/src/main/java/org/apache/kafka/streams/errors/StreamsException.java
index 6247886..b90f11f 100644
--- a/streams/src/main/java/org/apache/kafka/streams/errors/StreamsException.java
+++ b/streams/src/main/java/org/apache/kafka/streams/errors/StreamsException.java
@@ -21,20 +21,22 @@ import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.annotation.InterfaceStability;
 
 /**
- * StreamsException is the top-level exception type generated by Kafka Streams.
+ * {@link StreamsException} is the top-level exception type generated by Kafka Streams.
  */
 @InterfaceStability.Unstable
 public class StreamsException extends KafkaException {
 
-    public StreamsException(String s) {
-        super(s);
+    private final static long serialVersionUID = 1L;
+
+    public StreamsException(final String message) {
+        super(message);
     }
 
-    public StreamsException(String s, Throwable throwable) {
-        super(s, throwable);
+    public StreamsException(final String message, final Throwable throwable) {
+        super(message, throwable);
     }
 
-    public StreamsException(Throwable throwable) {
+    public StreamsException(final Throwable throwable) {
         super(throwable);
     }
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/10b330cf/streams/src/main/java/org/apache/kafka/streams/errors/TaskAssignmentException.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/errors/TaskAssignmentException.java
b/streams/src/main/java/org/apache/kafka/streams/errors/TaskAssignmentException.java
index 3936ef5..8c06b5b 100644
--- a/streams/src/main/java/org/apache/kafka/streams/errors/TaskAssignmentException.java
+++ b/streams/src/main/java/org/apache/kafka/streams/errors/TaskAssignmentException.java
@@ -17,21 +17,23 @@
 package org.apache.kafka.streams.errors;
 
 /**
- * Indicates a run time error incurred while trying to assign stream tasks to threads
+ * Indicates a run time error incurred while trying to assign
+ * {@link org.apache.kafka.streams.processor.internals.StreamTask stream tasks} to
+ * {@link org.apache.kafka.streams.processor.internals.StreamThread threads}.
  */
 public class TaskAssignmentException extends StreamsException {
 
     private final static long serialVersionUID = 1L;
 
-    public TaskAssignmentException(String s) {
-        super(s);
+    public TaskAssignmentException(final String message) {
+        super(message);
     }
 
-    public TaskAssignmentException(String s, Throwable throwable) {
-        super(s, throwable);
+    public TaskAssignmentException(final String message, final Throwable throwable) {
+        super(message, throwable);
     }
 
-    public TaskAssignmentException(Throwable throwable) {
+    public TaskAssignmentException(final Throwable throwable) {
         super(throwable);
     }
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/10b330cf/streams/src/main/java/org/apache/kafka/streams/errors/TaskIdFormatException.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/errors/TaskIdFormatException.java
b/streams/src/main/java/org/apache/kafka/streams/errors/TaskIdFormatException.java
index 576b972..20d739f 100644
--- a/streams/src/main/java/org/apache/kafka/streams/errors/TaskIdFormatException.java
+++ b/streams/src/main/java/org/apache/kafka/streams/errors/TaskIdFormatException.java
@@ -17,14 +17,28 @@
 
 package org.apache.kafka.streams.errors;
 
+import org.apache.kafka.common.annotation.InterfaceStability;
+
 /**
- * Indicates a run time error incurred while trying parse the task id from the read string
+ * Indicates a run time error incurred while trying parse the {@link org.apache.kafka.streams.processor.TaskId
task id}
+ * from the read string.
+ *
+ * @see org.apache.kafka.streams.processor.internals.StreamTask
  */
+@InterfaceStability.Unstable
 public class TaskIdFormatException extends StreamsException {
 
     private static final long serialVersionUID = 1L;
 
-    public TaskIdFormatException(String taskString) {
-        super("Task id cannot be parsed correctly" + (taskString == null ? "" : " from "
+ taskString));
+    public TaskIdFormatException(final String message) {
+        super("Task id cannot be parsed correctly" + (message == null ? "" : " from " + message));
+    }
+
+    public TaskIdFormatException(final String message, final Throwable throwable) {
+        super("Task id cannot be parsed correctly" + (message == null ? "" : " from " + message),
throwable);
+    }
+
+    public TaskIdFormatException(final Throwable throwable) {
+        super(throwable);
     }
 }

http://git-wip-us.apache.org/repos/asf/kafka/blob/10b330cf/streams/src/main/java/org/apache/kafka/streams/errors/TopologyBuilderException.java
----------------------------------------------------------------------
diff --git a/streams/src/main/java/org/apache/kafka/streams/errors/TopologyBuilderException.java
b/streams/src/main/java/org/apache/kafka/streams/errors/TopologyBuilderException.java
index 8745693..0789460 100644
--- a/streams/src/main/java/org/apache/kafka/streams/errors/TopologyBuilderException.java
+++ b/streams/src/main/java/org/apache/kafka/streams/errors/TopologyBuilderException.java
@@ -17,14 +17,26 @@
 
 package org.apache.kafka.streams.errors;
 
+import org.apache.kafka.common.annotation.InterfaceStability;
+
 /**
- * Indicates a pre-run time error incurred while parsing the builder to construct the processor
topology
+ * Indicates a pre-run time error incurred while parsing the {@link org.apache.kafka.streams.processor.TopologyBuilder
+ * builder} to construct the {@link org.apache.kafka.streams.processor.internals.ProcessorTopology
processor topology}.
  */
+@InterfaceStability.Unstable
 public class TopologyBuilderException extends StreamsException {
 
     private static final long serialVersionUID = 1L;
 
-    public TopologyBuilderException(String message) {
+    public TopologyBuilderException(final String message) {
         super("Invalid topology building" + (message == null ? "" : ": " + message));
     }
+
+    public TopologyBuilderException(final String message, final Throwable throwable) {
+        super("Invalid topology building" + (message == null ? "" : ": " + message), throwable);
+    }
+
+    public TopologyBuilderException(final Throwable throwable) {
+        super(throwable);
+    }
 }


Mime
View raw message