Fix some issues reported by Sonar
diff --git a/src/main/java/org/apache/sling/distribution/journal/kafka/CLSwitch.java b/src/main/java/org/apache/sling/distribution/journal/kafka/CLSwitch.java
new file mode 100644
index 0000000..cd06b7a
--- /dev/null
+++ b/src/main/java/org/apache/sling/distribution/journal/kafka/CLSwitch.java
@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.sling.distribution.journal.kafka;
+
+import java.io.Closeable;
+
+public class CLSwitch implements Closeable {
+ private ClassLoader oldClassloader;
+
+ public CLSwitch(ClassLoader classloader) {
+ oldClassloader = Thread.currentThread().getContextClassLoader();
+ Thread.currentThread().setContextClassLoader(classloader);
+ }
+
+ @Override
+ public void close() {
+ Thread.currentThread().setContextClassLoader(oldClassloader);
+ }
+
+}
diff --git a/src/main/java/org/apache/sling/distribution/journal/kafka/KafkaClientProvider.java b/src/main/java/org/apache/sling/distribution/journal/kafka/KafkaClientProvider.java
index 78046c0..61c8173 100644
--- a/src/main/java/org/apache/sling/distribution/journal/kafka/KafkaClientProvider.java
+++ b/src/main/java/org/apache/sling/distribution/journal/kafka/KafkaClientProvider.java
@@ -185,11 +185,10 @@
@Override
public long retrieveOffset(String topicName, Reset reset) {
- try (KafkaConsumer<String, String> consumer = createConsumer(StringDeserializer.class, reset)) {;
+ try (KafkaConsumer<String, String> consumer = createConsumer(StringDeserializer.class, reset)) {
TopicPartition topicPartition = new TopicPartition(topicName, PARTITION);
Map<TopicPartition, Long> offsets = getOffsets(reset, consumer, topicPartition);
- Long offset = offsets.get(topicPartition);
- return offset;
+ return offsets.get(topicPartition);
}
}
@@ -208,12 +207,8 @@
protected <T> KafkaConsumer<String, T> createConsumer(Class<? extends Deserializer<?>> deserializer, Reset reset) {
String groupId = UUID.randomUUID().toString();
- ClassLoader oldClassloader = Thread.currentThread().getContextClassLoader();
- Thread.currentThread().setContextClassLoader(KafkaConsumer.class.getClassLoader());
- try {
+ try (CLSwitch switcher = new CLSwitch(KafkaConsumer.class.getClassLoader())) {
return new KafkaConsumer<>(consumerConfig(deserializer, groupId, reset));
- } finally {
- Thread.currentThread().setContextClassLoader(oldClassloader);
}
}
