1.Remove redundant exception handling functions in KafkaExceptionHandler.2.Add Kafka Topic JMX Filter template.

Author: tinawenqiao

JMXFilterTemplate/KafkaTopicFilter.yml

@@ -0,0 +1,57 @@
+filters:
+    - include:
+        domain: kafka.server
+        bean: kafka.server:type=BrokerTopicMetrics,name=BytesInPerSec,topic={topicname}
+        attribute:
+            Count:
+                metric_type: KAFKA_TOPIC_PERF
+                alias: BytesInPerSec
+    - include:
+        domain: kafka.server
+        bean: kafka.server:type=BrokerTopicMetrics,name=BytesOutPerSec,topic={topicname}
+        attribute:
+            Count:
+                metric_type: KAFKA_TOPIC_PERF
+                alias: BytesOutPerSec
+    - include:
+        domain: kafka.server
+        bean: kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec,topic={topicname}
+        attribute:
+            Count:
+                metric_type: KAFKA_TOPIC_PERF
+                alias: MessagesInPerSec
+    - include:
+        domain: kafka.server
+        bean: kafka.server:type=BrokerTopicMetrics,name=BytesRejectedPerSec,topic={topicname}
+        attribute:
+            Count:
+                metric_type: KAFKA_TOPIC_PERF
+                alias: BytesRejectedPerSec
+    - include:
+        domain: kafka.server
+        bean: kafka.server:type=BrokerTopicMetrics,name=FailedFetchRequestsPerSec,topic={topicname}
+        attribute:
+            Count:
+                metric_type: KAFKA_TOPIC_PERF
+                alias: FailedFetchRequestsPerSec
+    - include:
+        domain: kafka.server
+        bean: kafka.server:type=BrokerTopicMetrics,name=FailedProduceRequestsPerSec,topic={topicname}
+        attribute:
+            Count:
+                metric_type: KAFKA_TOPIC_PERF
+                alias: FailedProduceRequestsPerSec
+    - include:
+        domain: kafka.server
+        bean: kafka.server:type=BrokerTopicMetrics,name=TotalFetchRequestsPerSec,topic={topicname}
+        attribute:
+            Count:
+                metric_type: KAFKA_TOPIC_PERF
+                alias: TotalFetchRequestsPerSec
+    - include:
+            domain: kafka.server
+            bean: kafka.server:type=BrokerTopicMetrics,name=TotalProduceRequestsPerSec,topic={topicname}
+            attribute:
+                Count:
+                    metric_type: KAFKA_TOPIC_PERF
+                    alias: TotalProduceRequestsPerSec

src/main/java/org/gnuhpc/bigdata/exception/KafkaExceptionHandler.java

@@ -14,17 +14,6 @@
 @Log4j
 @RestControllerAdvice
 public class KafkaExceptionHandler {
-    @ExceptionHandler(Exception.class)
-    public RestErrorResponse handleException(Exception ex){
-        RestErrorResponse.Builder builder = new RestErrorResponse.Builder();
-        RestErrorResponse response = builder
-                .setCode(KafkaErrorCode.UNKNOWN.ordinal())
-                .setMessage("Default Exception happened!")
-                .setDeveloperMessage(ex.getMessage())
-                .setStatus(HttpStatus.SERVICE_UNAVAILABLE).build();
-        return response;
-    }
-
     @ExceptionHandler(ApiException.class)
     public RestErrorResponse kafkaApiException(ApiException ex) {
         RestErrorResponse.Builder responseBuilder = new RestErrorResponse.Builder();
@@ -34,43 +23,4 @@ public RestErrorResponse kafkaApiException(ApiException ex) {
                 .setDeveloperMessage(ex.getMessage())
                 .build();
     }
-
-    @ExceptionHandler(RuntimeException.class)
-    public RestErrorResponse runtimeException(RuntimeException ex){
-        RestErrorResponse.Builder responseBuilder = new RestErrorResponse.Builder();
-        return responseBuilder.setStatus(HttpStatus.SERVICE_UNAVAILABLE)
-                .setCode(KafkaErrorCode.UNKNOWN.ordinal())
-                .setMessage("Runtime Exception happened!")
-                .setDeveloperMessage(ex.getMessage())
-                .build();
-    }
-
-    /*
-    @ExceptionHandler(ConstraintViolationException.class)
-    public RestErrorResponse constraintViolationException(ConstraintViolationException ex){
-        StringBuilder message = new StringBuilder();
-        Set<ConstraintViolation<?>> violations = ex.getConstraintViolations();
-        for (ConstraintViolation<?> violation : violations) {
-            message.append(violation.getMessage().concat(";"));
-        }
-
-        RestErrorResponse.Builder responseBuilder = new RestErrorResponse.Builder();
-        return responseBuilder.setStatus(HttpStatus.SERVICE_UNAVAILABLE)
-                .setCode(KafkaErrorCode.UNKNOWN_TOPIC_OR_PARTITION.ordinal())
-                .setMessage("Constraint  Violation Exception happened!")
-                .setMessage(message.toString().substring(0,message.length()-1))
-                .setDeveloperMessage(ex.getMessage())
-                .build();
-    }
-    */
-
-    @ExceptionHandler(ResourceNotFoundException.class)
-    public RestErrorResponse serviceNotAvailableException(ServiceNotAvailableException ex){
-        RestErrorResponse.Builder responseBuilder = new RestErrorResponse.Builder();
-        return responseBuilder.setStatus(HttpStatus.SERVICE_UNAVAILABLE)
-                .setCode(KafkaErrorCode.SERVICE_DOWN.ordinal())
-                .setMessage("Service not Available happened: " + ex)
-                .setDeveloperMessage(ex.getMessage())
-                .build();
-    }
 }