Enable spotbugs check in the module pulsar-common (apache#9016)

Enable spotbugs check in the module pulsar-common.

pulsar-common/pom.xml

@@ -224,8 +224,23 @@
         </execution>
       </executions>
     </plugin>
-
-
+    <plugin>
+      <groupId>com.github.spotbugs</groupId>
+      <artifactId>spotbugs-maven-plugin</artifactId>
+      <version>${spotbugs-maven-plugin.version}</version>
+      <executions>
+        <execution>
+          <id>spotbugs</id>
+          <phase>verify</phase>
+          <goals>
+            <goal>check</goal>
+          </goals>
+        </execution>
+      </executions>
+      <configuration>
+        <excludeFilterFile>${basedir}/src/main/resources/findbugsExclude.xml</excludeFilterFile>
+      </configuration>
+    </plugin>
   </plugins>
   </build>
 </project>

pulsar-common/src/main/java/org/apache/pulsar/client/api/url/DataURLStreamHandler.java

@@ -27,6 +27,7 @@
 import java.net.URL;
 import java.net.URLConnection;
 import java.net.URLStreamHandler;
+import java.nio.charset.StandardCharsets;
 import java.util.Base64;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
@@ -76,7 +77,7 @@ public void connect() throws IOException {
 
                 if (matcher.group("base64") == null) {
                     // Support Urlencode but not decode here because already decoded by URI class.
-                    this.data = matcher.group("data").getBytes();
+                    this.data = matcher.group("data").getBytes(StandardCharsets.UTF_8);
                 } else {
                     this.data = Base64.getDecoder().decode(matcher.group("data"));
                 }

pulsar-common/src/main/java/org/apache/pulsar/common/api/raw/RawMessageImpl.java

@@ -22,6 +22,7 @@
 import io.netty.buffer.Unpooled;
 import io.netty.util.Recycler;
 import io.netty.util.Recycler.Handle;
+import java.nio.charset.StandardCharsets;
 import java.util.Base64;
 import java.util.Collections;
 import java.util.Map;
@@ -155,7 +156,7 @@ public Optional<ByteBuf> getKeyBytes() {
             if (hasBase64EncodedKey()) {
                 return Optional.of(Unpooled.wrappedBuffer(Base64.getDecoder().decode(getKey().get())));
             } else {
-                return Optional.of(Unpooled.wrappedBuffer(getKey().get().getBytes()));
+                return Optional.of(Unpooled.wrappedBuffer(getKey().get().getBytes(StandardCharsets.UTF_8)));
             }
         }
         return Optional.empty();

pulsar-common/src/main/java/org/apache/pulsar/common/nar/FileUtils.java

@@ -181,7 +181,8 @@ public static void deleteFilesInDirectory(
                 }
                 if (ingestFile.isDirectory() && recurse) {
                     FileUtils.deleteFilesInDirectory(ingestFile, filter, logger, recurse, deleteEmptyDirectories);
-                    if (deleteEmptyDirectories && ingestFile.list().length == 0) {
+                    String[] ingestFileList = ingestFile.list();
+                    if (deleteEmptyDirectories && ingestFileList != null && ingestFileList.length == 0) {
                         FileUtils.deleteFile(ingestFile, logger, 3);
                     }
                 }

pulsar-common/src/main/java/org/apache/pulsar/common/nar/NarClassLoader.java

@@ -26,18 +26,25 @@
 import java.io.BufferedReader;
 import java.io.File;
 import java.io.FileFilter;
+import java.io.FileInputStream;
 import java.io.FileReader;
 import java.io.IOException;
+import java.io.InputStreamReader;
 import java.net.URL;
 import java.net.URLClassLoader;
+import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
+import java.security.AccessController;
+import java.security.PrivilegedAction;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Comparator;
 import java.util.List;
 import java.util.Set;
+
+import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 
 /**
@@ -151,11 +158,13 @@ public static NarClassLoader getFromArchive(File narPath, Set<String> additional
                                                 String narExtractionDirectory)
         throws IOException {
         File unpacked = NarUnpacker.unpackNar(narPath, getNarExtractionDirectory(narExtractionDirectory));
-        try {
-            return new NarClassLoader(unpacked, additionalJars, parent);
-        } catch (ClassNotFoundException | NoClassDefFoundError e) {
-            throw new IOException(e);
-        }
+        return AccessController.doPrivileged(new PrivilegedAction<NarClassLoader>() {
+            @SneakyThrows
+            @Override
+            public NarClassLoader run() {
+                return new NarClassLoader(unpacked, additionalJars, parent);
+            }
+        });
     }
 
     private static File getNarExtractionDirectory(String configuredDirectory) {
@@ -208,15 +217,16 @@ public File getWorkingDirectory() {
      */
     public String getServiceDefinition(String serviceName) throws IOException {
         String serviceDefPath = narWorkingDirectory + "/META-INF/services/" + serviceName;
-        return new String(Files.readAllBytes(Paths.get(serviceDefPath)));
+        return new String(Files.readAllBytes(Paths.get(serviceDefPath)), StandardCharsets.UTF_8);
     }
 
     public List<String> getServiceImplementation(String serviceName) throws IOException {
         List<String> impls = new ArrayList<>();
 
         String serviceDefPath = narWorkingDirectory + "/META-INF/services/" + serviceName;
 
-        try (BufferedReader reader = new BufferedReader(new FileReader(serviceDefPath))) {
+        try (BufferedReader reader = new BufferedReader(
+            new InputStreamReader(new FileInputStream(serviceDefPath), StandardCharsets.UTF_8))) {
             String line;
             while ((line = reader.readLine()) != null) {
                 line = line.trim();

pulsar-common/src/main/java/org/apache/pulsar/common/net/ServiceURI.java

@@ -26,6 +26,7 @@
 import com.google.common.base.Strings;
 import java.net.URI;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 import java.util.stream.Collectors;
 import lombok.AccessLevel;
@@ -220,7 +221,8 @@ private static int getServicePort(String serviceName, String[] serviceInfos) {
                 } else if (serviceInfos.length == 1 && serviceInfos[0].toLowerCase().equals(SSL_SERVICE)) {
                     port = BINARY_TLS_PORT;
                 } else {
-                    throw new IllegalArgumentException("Invalid pulsar service : " + serviceName + "+" + serviceInfos);
+                    throw new IllegalArgumentException("Invalid pulsar service : " + serviceName + "+"
+                        + Arrays.toString(serviceInfos));
                 }
                 break;
             case HTTP_SERVICE:

pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/BrokerStatus.java

@@ -19,10 +19,12 @@
 package org.apache.pulsar.common.policies.data;
 
 import com.google.common.collect.ComparisonChain;
+import lombok.EqualsAndHashCode;
 
 /**
  * Information about the broker status.
  */
+@EqualsAndHashCode
 public class BrokerStatus implements Comparable<BrokerStatus> {
     private String brokerAddress;
     private boolean active;

pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/FunctionStats.java

@@ -27,6 +27,7 @@
 import java.util.List;
 import java.util.Map;
 import lombok.Data;
+import lombok.EqualsAndHashCode;
 import org.apache.pulsar.common.util.ObjectMapperFactory;
 
 /**
@@ -120,6 +121,7 @@ public static class FunctionInstanceStatsDataBase {
         /**
          * Function instance statistics data.
          */
+        @EqualsAndHashCode(callSuper = true)
         @Data
         @JsonInclude(JsonInclude.Include.ALWAYS)
         @JsonPropertyOrder({ "receivedTotal", "processedSuccessfullyTotal", "systemExceptionsTotal",

pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/OffloadPolicies.java

@@ -60,7 +60,7 @@ public class OffloadPolicies implements Serializable {
     public final static int DEFAULT_READ_BUFFER_SIZE_IN_BYTES = 1024 * 1024;      // 1MB
     public final static int DEFAULT_OFFLOAD_MAX_THREADS = 2;
     public final static int DEFAULT_OFFLOAD_MAX_PREFETCH_ROUNDS = 1;
-    public final static String[] DRIVER_NAMES = {
+    final static String[] DRIVER_NAMES = {
             "S3", "aws-s3", "google-cloud-storage", "filesystem", "azureblob", "aliyun-oss"
     };
     public final static String DEFAULT_OFFLOADER_DIRECTORY = "./offloaders";

pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/PersistentOfflineTopicStats.java

@@ -72,7 +72,7 @@ public void reset() {
     /**
      * Details about a cursor.
      */
-    public class CursorDetails {
+    public static class CursorDetails {
         public long cursorBacklog;
         public long cursorLedgerId;
 
@@ -93,7 +93,7 @@ public void addLedgerDetails(long entries, long timestamp, long size, long ledge
     /**
      * Details about a ledger.
      */
-    public class LedgerDetails {
+    public static class LedgerDetails {
         public long entries;
         public long timestamp;
         public long size;

pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/Policies.java

@@ -240,15 +240,14 @@ public String toString() {
                 .add("offload_policies", offload_policies).toString();
     }
 
-
     private static final long MAX_BUNDLES = ((long) 1) << 32;
 
     public static BundlesData getBundles(int numBundles) {
-        if (numBundles <= 0 || numBundles > MAX_BUNDLES) {
+        if (numBundles <= 0) {
             throw new RestException(Status.BAD_REQUEST,
-                    "Invalid number of bundles. Number of numbles has to be in the range of (0, 2^32].");
+                "Invalid number of bundles. Number of numbles has to be in the range of (0, 2^32].");
         }
-        Long maxVal = ((long) 1) << 32;
+        Long maxVal = MAX_BUNDLES;
         Long segSize = maxVal / numBundles;
         List<String> partitions = Lists.newArrayList();
         partitions.add(String.format("0x%08x", 0L));

pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Markers.java

@@ -99,10 +99,8 @@ public static ByteBuf newReplicatedSubscriptionsSnapshotRequest(String snapshotI
     public static ReplicatedSubscriptionsSnapshotRequest parseReplicatedSubscriptionsSnapshotRequest(ByteBuf payload)
             throws IOException {
         ByteBufCodedInputStream inStream = ByteBufCodedInputStream.get(payload);
-        ReplicatedSubscriptionsSnapshotRequest.Builder builder = null;
-
+        ReplicatedSubscriptionsSnapshotRequest.Builder builder = ReplicatedSubscriptionsSnapshotRequest.newBuilder();
         try {
-            builder = ReplicatedSubscriptionsSnapshotRequest.newBuilder();
             return builder.mergeFrom(inStream, null).build();
         } finally {
             builder.recycle();
@@ -148,10 +146,8 @@ public static ByteBuf newReplicatedSubscriptionsSnapshotResponse(String snapshot
     public static ReplicatedSubscriptionsSnapshotResponse parseReplicatedSubscriptionsSnapshotResponse(ByteBuf payload)
             throws IOException {
         ByteBufCodedInputStream inStream = ByteBufCodedInputStream.get(payload);
-        ReplicatedSubscriptionsSnapshotResponse.Builder builder = null;
-
+        ReplicatedSubscriptionsSnapshotResponse.Builder builder = ReplicatedSubscriptionsSnapshotResponse.newBuilder();
         try {
-            builder = ReplicatedSubscriptionsSnapshotResponse.newBuilder();
             return builder.mergeFrom(inStream, null).build();
         } finally {
             builder.recycle();
@@ -198,10 +194,8 @@ public static ByteBuf newReplicatedSubscriptionsSnapshot(String snapshotId, Stri
     public static ReplicatedSubscriptionsSnapshot parseReplicatedSubscriptionsSnapshot(ByteBuf payload)
             throws IOException {
         ByteBufCodedInputStream inStream = ByteBufCodedInputStream.get(payload);
-        ReplicatedSubscriptionsSnapshot.Builder builder = null;
-
+        ReplicatedSubscriptionsSnapshot.Builder builder = ReplicatedSubscriptionsSnapshot.newBuilder();
         try {
-            builder = ReplicatedSubscriptionsSnapshot.newBuilder();
             return builder.mergeFrom(inStream, null).build();
         } finally {
             builder.recycle();
@@ -243,10 +237,8 @@ public static ByteBuf newReplicatedSubscriptionsUpdate(String subscriptionName,
     public static ReplicatedSubscriptionsUpdate parseReplicatedSubscriptionsUpdate(ByteBuf payload)
             throws IOException {
         ByteBufCodedInputStream inStream = ByteBufCodedInputStream.get(payload);
-        ReplicatedSubscriptionsUpdate.Builder builder = null;
-
+        ReplicatedSubscriptionsUpdate.Builder builder = ReplicatedSubscriptionsUpdate.newBuilder();
         try {
-            builder = ReplicatedSubscriptionsUpdate.newBuilder();
             return builder.mergeFrom(inStream, null).build();
         } finally {
             builder.recycle();
@@ -280,11 +272,8 @@ public static ByteBuf newTxnAbortMarker(long sequenceId, long txnMostBits,
 
     public static PulsarMarkers.TxnCommitMarker parseCommitMarker(ByteBuf payload) throws IOException {
         ByteBufCodedInputStream inStream = ByteBufCodedInputStream.get(payload);
-
-        PulsarMarkers.TxnCommitMarker.Builder builder = null;
-
+        PulsarMarkers.TxnCommitMarker.Builder builder = PulsarMarkers.TxnCommitMarker.newBuilder();
         try {
-            builder = PulsarMarkers.TxnCommitMarker.newBuilder();
             return builder.mergeFrom(inStream, null).build();
         } finally {
             builder.recycle();

pulsar-common/src/main/java/org/apache/pulsar/common/protocol/PulsarDecoder.java

@@ -486,6 +486,8 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
                 handleEndTxnOnSubscriptionResponse(cmd.getEndTxnOnSubscriptionResponse());
                 cmd.getEndTxnOnSubscriptionResponse().recycle();
                 break;
+            default:
+                break;
             }
         } finally {
             if (cmdBuilder != null) {

pulsar-common/src/main/java/org/apache/pulsar/common/protocol/schema/SchemaInfoUtil.java

@@ -18,6 +18,7 @@
  */
 package org.apache.pulsar.common.protocol.schema;
 
+import java.nio.charset.StandardCharsets;
 import java.util.Collections;
 import java.util.TreeMap;
 import lombok.experimental.UtilityClass;
@@ -61,7 +62,7 @@ public static SchemaInfo newSchemaInfo(Schema schema) {
     public static SchemaInfo newSchemaInfo(String name, GetSchemaResponse schema) {
         SchemaInfo si = new SchemaInfo();
         si.setName(name);
-        si.setSchema(schema.getData().getBytes());
+        si.setSchema(schema.getData().getBytes(StandardCharsets.UTF_8));
         si.setType(schema.getType());
         si.setProperties(schema.getProperties());
         return si;

pulsar-common/src/main/java/org/apache/pulsar/common/stats/Rate.java

@@ -19,6 +19,7 @@
 package org.apache.pulsar.common.stats;
 
 import static com.google.common.base.Preconditions.checkArgument;
+import java.math.BigDecimal;
 import java.util.concurrent.atomic.LongAdder;
 
 /**
@@ -29,7 +30,7 @@ public class Rate {
     private final LongAdder countAdder = new LongAdder();
 
     // Computed stats
-    private long count = 0;
+    private long count = 0L;
     private double rate = 0.0d;
     private double valueRate = 0.0d;
     private double averageValue = 0.0d;
@@ -60,7 +61,7 @@ public void calculateRate(double period) {
 
         count = countAdder.sumThenReset();
         long sum = valueAdder.sumThenReset();
-        averageValue = count != 0 ? sum / count : 0.0d;
+        averageValue = count != 0 ? Long.valueOf(sum).doubleValue() / Long.valueOf(count).doubleValue() : 0.0d;
         rate = count / period;
         valueRate = sum / period;
     }

pulsar-common/src/main/java/org/apache/pulsar/common/util/ClassLoaderUtils.java

@@ -23,6 +23,8 @@
 import java.net.URL;
 import java.net.URLClassLoader;
 import java.nio.file.Path;
+import java.security.AccessController;
+import java.security.PrivilegedAction;
 
 /**
  * Helper methods wrt Classloading.
@@ -37,7 +39,8 @@ public class ClassLoaderUtils {
      */
     public static ClassLoader loadJar(File jar) throws MalformedURLException {
         java.net.URL url = jar.toURI().toURL();
-        return new URLClassLoader(new URL[]{url});
+        return AccessController.doPrivileged(
+            (PrivilegedAction<URLClassLoader>) () -> new URLClassLoader(new URL[]{url}));
     }
 
     public static ClassLoader extractClassLoader(Path archivePath, File packageFile) throws Exception {

pulsar-common/src/main/java/org/apache/pulsar/common/util/NamespaceBundleStatsComparator.java

@@ -18,14 +18,15 @@
  */
 package org.apache.pulsar.common.util;
 
+import java.io.Serializable;
 import java.util.Comparator;
 import java.util.Map;
 import org.apache.pulsar.policies.data.loadbalancer.NamespaceBundleStats;
 import org.apache.pulsar.policies.data.loadbalancer.SystemResourceUsage.ResourceType;
 
 /**
  */
-public class NamespaceBundleStatsComparator implements Comparator<String> {
+public class NamespaceBundleStatsComparator implements Comparator<String>, Serializable {
     Map<String, NamespaceBundleStats> map;
     ResourceType resType;
 

pulsar-common/src/main/java/org/apache/pulsar/common/util/RestException.java

@@ -39,7 +39,9 @@ public static String getExceptionData(Throwable t) {
         }
         writer.append("Stacktrace:\n\n");
 
-        t.printStackTrace(new PrintWriter(writer));
+        if (t != null) {
+            t.printStackTrace(new PrintWriter(writer));
+        }
         return writer.toString();
     }