apache · sijie · Jan 13, 2021 · Dec 7, 2020 · Dec 11, 2020 · Dec 11, 2020
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java
@@ -237,7 +237,7 @@ private CompletableFuture<NamespaceBundle> getFullBundleAsync(NamespaceName fqnn
     /**
      * Return the URL of the broker who's owning a particular service unit in asynchronous way.
      *
-     * If the service unit is not owned, return a CompletableFuture with empty optional
+     * If the service unit is not owned, return a CompletableFuture with empty optional.
      */
     public CompletableFuture<Optional<URL>> getWebServiceUrlAsync(ServiceUnitId suName, LookupOptions options)
             throws Exception {

diff --git a/pulsar-functions/api-java/pom.xml b/pulsar-functions/api-java/pom.xml
@@ -51,4 +51,26 @@
 
   </dependencies>
 
+  <build>
+    <plugins>
+      <plugin>
+        <groupId>com.github.spotbugs</groupId>
+        <artifactId>spotbugs-maven-plugin</artifactId>
+        <version>${spotbugs-maven-plugin.version}</version>
+        <configuration>
+          <excludeFilterFile>${basedir}/src/main/resources/findbugsExclude.xml</excludeFilterFile>
+        </configuration>
+        <executions>
+          <execution>
+            <id>check</id>
+            <phase>verify</phase>
+            <goals>
+              <goal>check</goal>
+            </goals>
+          </execution>
+        </executions>
+      </plugin>
+    </plugins>
+  </build>
+
 </project>
diff --git a/pulsar-functions/api-java/src/main/resources/findbugsExclude.xml b/pulsar-functions/api-java/src/main/resources/findbugsExclude.xml
@@ -0,0 +1,22 @@
+<!--
+
+    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.
+
+-->
+<FindBugsFilter>
+</FindBugsFilter>
diff --git a/pulsar-functions/instance/pom.xml b/pulsar-functions/instance/pom.xml
@@ -200,6 +200,23 @@
           </execution>
         </executions>
       </plugin>
+      <plugin>
+        <groupId>com.github.spotbugs</groupId>
+        <artifactId>spotbugs-maven-plugin</artifactId>
+        <version>${spotbugs-maven-plugin.version}</version>
+        <configuration>
+          <excludeFilterFile>${basedir}/src/main/resources/findbugsExclude.xml</excludeFilterFile>
+        </configuration>
+        <executions>
+          <execution>
+            <id>check</id>
+            <phase>verify</phase>
+            <goals>
+              <goal>check</goal>
+            </goals>
+          </execution>
+        </executions>
+      </plugin>
     </plugins>
   </build>
 

diff --git a/...-functions/instance/src/main/java/org/apache/pulsar/functions/instance/InstanceUtils.java b/...-functions/instance/src/main/java/org/apache/pulsar/functions/instance/InstanceUtils.java
@@ -140,6 +140,8 @@ public static Map<String, String> getProperties(Function.FunctionDetails.Compone
             case SINK:
                 properties.put("application", "pulsar-sink");
                 break;
+            default:
+                throw new IllegalArgumentException("Not support component type");
         }
         properties.put("id", fullyQualifiedName);
         properties.put("instance_id", String.valueOf(instanceId));

diff --git a/...ar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/LogAppender.java b/...ar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/LogAppender.java
@@ -18,6 +18,7 @@
  */
 package org.apache.pulsar.functions.instance;
 
+import java.nio.charset.StandardCharsets;
 import org.apache.logging.log4j.core.*;
 import org.apache.pulsar.client.api.CompressionType;
 import org.apache.pulsar.client.api.Producer;
@@ -47,7 +48,7 @@ public LogAppender(PulsarClient pulsarClient, String logTopic, String fqn) {
     @Override
     public void append(LogEvent logEvent) {
         producer.newMessage()
-                .value(logEvent.getMessage().getFormattedMessage().getBytes())
+                .value(logEvent.getMessage().getFormattedMessage().getBytes(StandardCharsets.UTF_8))
                 .property("loglevel", logEvent.getLevel().name())
                 .sendAsync();
     }

diff --git a/...ce/src/main/java/org/apache/pulsar/functions/instance/state/BKStateStoreProviderImpl.java b/...ce/src/main/java/org/apache/pulsar/functions/instance/state/BKStateStoreProviderImpl.java
@@ -119,8 +119,13 @@ private void createStateTable(String stateStorageServiceUrl,
                 } catch (NamespaceNotFoundException nnfe) {
                     try {
                         result(storageAdminClient.createNamespace(tableNs, NamespaceConfiguration.newBuilder()
-                            .setDefaultStreamConf(streamConf)
-                            .build()));
+                                .setDefaultStreamConf(streamConf)
+                                .build()));
+                    } catch (Exception e) {
+                        // there might be two clients conflicting at creating table, so let's retrieve the table again
+                        // to make sure the table is created.
+                    }
+                    try {
                         result(storageAdminClient.createStream(tableNs, tableName, streamConf));
                     } catch (Exception e) {
                         // there might be two clients conflicting at creating table, so let's retrieve the table again

diff --git a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/sink/PulsarSink.java
@@ -19,9 +19,12 @@
 package org.apache.pulsar.functions.sink;
 
 import com.google.common.annotations.VisibleForTesting;
+
+import java.nio.charset.StandardCharsets;
 import lombok.Builder;
 import lombok.Data;
 import lombok.extern.slf4j.Slf4j;
+import lombok.val;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.pulsar.client.api.BatcherBuilder;
 import org.apache.pulsar.client.api.CompressionType;
@@ -126,7 +129,6 @@ public Producer<T> createProducer(PulsarClient client, String topic, String prod
                     builder.maxPendingMessagesAcrossPartitions(producerConfig.getMaxPendingMessagesAcrossPartitions());
                 }
                 if (producerConfig.getCryptoConfig() != null) {
-                    CryptoConfig cryptoConfig = producerConfig.getCryptoConfig();
                     builder.cryptoKeyReader(crypto.keyReader);
                     builder.cryptoFailureAction(crypto.failureAction);
                     for (String encryptionKeyName : crypto.getEncryptionKeys()) {
@@ -366,7 +368,7 @@ public void write(Record<T> record) {
             // forward user properties to sink-topic
             msg.property("__pfn_input_topic__", pulsarRecord.getTopicName().get())
                .property("__pfn_input_msg_id__",
-                         new String(Base64.getEncoder().encode(pulsarRecord.getMessageId().toByteArray())));
+                         new String(Base64.getEncoder().encode(pulsarRecord.getMessageId().toByteArray()), StandardCharsets.UTF_8));
         } else {
             // It is coming from some source
             Optional<Long> eventTime = sinkRecord.getSourceRecord().getEventTime();
@@ -424,9 +426,10 @@ Crypto initializeCrypto() throws ClassNotFoundException {
             Security.addProvider(new BouncyCastleProvider());
         }
 
+        final String[] encryptionKeys = cryptoConfig.getEncryptionKeys();
         Crypto.CryptoBuilder bldr = Crypto.builder()
                 .failureAction(cryptoConfig.getProducerCryptoFailureAction())
-                .encryptionKeys(cryptoConfig.getEncryptionKeys());
+                .encryptionKeys(encryptionKeys);
 
         bldr.keyReader(CryptoUtils.getCryptoKeyReaderInstance(
                 cryptoConfig.getCryptoKeyReaderClassName(), cryptoConfig.getCryptoKeyReaderConfig(), functionClassLoader));

diff --git a/pulsar-functions/instance/src/main/resources/findbugsExclude.xml b/pulsar-functions/instance/src/main/resources/findbugsExclude.xml
@@ -0,0 +1,71 @@
+<!--
+
+    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.
+
+-->
+<FindBugsFilter>
+    <!-- these public fields may be used in other modules -->
+    <Match>
+        <Class name="org.apache.pulsar.functions.instance.ContextImpl$MessageBuilderImpl"/>
+        <Bug pattern="RV_RETURN_VALUE_IGNORED_NO_SIDE_EFFECT"/>
+    </Match>
+    <Match>
+        <Class name="org.apache.pulsar.functions.instance.JavaInstanceRunnable"/>
+        <Bug pattern="IS2_INCONSISTENT_SYNC"/>
+    </Match>
+    <Match>
+        <Class name="org.apache.pulsar.functions.sink.PulsarSink"/>
+        <Bug pattern="RV_RETURN_VALUE_IGNORED_NO_SIDE_EFFECT"/>
+    </Match>
+    <Match>
+        <Class name="org.apache.pulsar.functions.sink.PulsarSink$PulsarSinkEffectivelyOnceProcessor"/>
+        <Bug pattern="RV_RETURN_VALUE_IGNORED_NO_SIDE_EFFECT"/>
+    </Match>
+    <Match>
+        <Class name="org.apache.pulsar.functions.instance.stats.ComponentStatsManager"/>
+        <Bug pattern="MS_MUTABLE_ARRAY"/>
+    </Match>
+    <Match>
+        <Class name="org.apache.pulsar.functions.instance.stats.ComponentStatsManager"/>
+        <Bug pattern="EI_EXPOSE_REP2"/>
+    </Match>
+    <Match>
+        <Class name="org.apache.pulsar.functions.instance.stats.ComponentStatsManager"/>
+        <Bug pattern="MS_PKGPROTECT"/>
+    </Match>
+    <Match>
+        <Class name="org.apache.pulsar.functions.sink.PulsarSink$Crypto$CryptoBuilder"/>
+        <Bug pattern="EI_EXPOSE_REP2"/>
+    </Match>
+    <Match>
+        <Class name="org.apache.pulsar.functions.source.batch.BatchSourceExecutor"/>
+        <Bug pattern="UC_USELESS_OBJECT"/>
+    </Match>
+    <Match>
+        <Class name="org.apache.pulsar.functions.source.PulsarSource"/>
+        <Bug pattern="SE_BAD_FIELD"/>
+    </Match>
+    <Match>
+        <Class name="org.apache.pulsar.functions.source.batch.BatchSourceExecutor"/>
+        <Bug pattern="RV_RETURN_VALUE_IGNORED_NO_SIDE_EFFECT"/>
+    </Match>
+    <Match>
+        <Class name="org.apache.pulsar.functions.source.SerDeSchema"/>
+        <Bug pattern="CN_IDIOM_NO_SUPER_CALL"/>
+    </Match>
+</FindBugsFilter>
diff --git a/pulsar-functions/java-examples/pom.xml b/pulsar-functions/java-examples/pom.xml
@@ -48,4 +48,26 @@
     </dependency>
   </dependencies>
 
+  <build>
+    <plugins>
+      <plugin>
+        <groupId>com.github.spotbugs</groupId>
+        <artifactId>spotbugs-maven-plugin</artifactId>
+        <version>${spotbugs-maven-plugin.version}</version>
+        <configuration>
+          <excludeFilterFile>${basedir}/src/main/resources/findbugsExclude.xml</excludeFilterFile>
+        </configuration>
+        <executions>
+          <execution>
+            <id>check</id>
+            <phase>verify</phase>
+            <goals>
+              <goal>check</goal>
+            </goals>
+          </execution>
+        </executions>
+      </plugin>
+    </plugins>
+  </build>
+
 </project>
diff --git a/...examples/src/main/java/org/apache/pulsar/functions/api/examples/AsyncContextFunction.java b/...examples/src/main/java/org/apache/pulsar/functions/api/examples/AsyncContextFunction.java
@@ -43,9 +43,8 @@ public CompletableFuture<Void> process(String input, Context context) {
             String inputTopics = context.getInputTopics().stream().collect(Collectors.joining(", "));
             String funcName = context.getFunctionName();
 
-            String logMessage = String
-                    .format("A message with value of \"%s\" has arrived on one of the following topics: %s\n",
-                            input, inputTopics);
+            String logMessage = String.format("A message with value of \"%s\" has arrived on " +
+                    "one of the following topics: %s %n", input, inputTopics);
             LOG.info(logMessage);
 
             String metricName = String.format("function-%s-messages-received", funcName);

diff --git a/...java-examples/src/main/java/org/apache/pulsar/functions/api/examples/ContextFunction.java b/...java-examples/src/main/java/org/apache/pulsar/functions/api/examples/ContextFunction.java
@@ -31,9 +31,8 @@ public Void process(String input, Context context) {
         String inputTopics = context.getInputTopics().stream().collect(Collectors.joining(", "));
         String functionName = context.getFunctionName();
 
-        String logMessage = String.format("A message with a value of \"%s\" has arrived on one of the following topics: %s\n",
-                input,
-                inputTopics);
+        String logMessage = String.format("A message with a value of \"%s\" has arrived on one of " +
+                "the following topics: %s %n", input, inputTopics);
 
         LOG.info(logMessage);
 

diff --git a/pulsar-functions/java-examples/src/main/resources/findbugsExclude.xml b/pulsar-functions/java-examples/src/main/resources/findbugsExclude.xml
@@ -0,0 +1,22 @@
+<!--
+
+    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.
+
+-->
+<FindBugsFilter>
+</FindBugsFilter>
diff --git a/pulsar-functions/localrun/pom.xml b/pulsar-functions/localrun/pom.xml
@@ -59,4 +59,26 @@
             <artifactId>grpc-all</artifactId>
         </dependency>
     </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>com.github.spotbugs</groupId>
+                <artifactId>spotbugs-maven-plugin</artifactId>
+                <version>${spotbugs-maven-plugin.version}</version>
+                <configuration>
+                    <excludeFilterFile>${basedir}/src/main/resources/findbugsExclude.xml</excludeFilterFile>
+                </configuration>
+                <executions>
+                    <execution>
+                        <id>check</id>
+                        <phase>verify</phase>
+                        <goals>
+                            <goal>check</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
 </project>
diff --git a/pulsar-functions/localrun/src/main/java/org/apache/pulsar/functions/LocalRunner.java b/pulsar-functions/localrun/src/main/java/org/apache/pulsar/functions/LocalRunner.java
@@ -24,6 +24,9 @@
 import com.google.gson.Gson;
 import com.google.gson.GsonBuilder;
 import com.google.gson.JsonParser;
+
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeoutException;
 import lombok.Builder;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.common.functions.FunctionConfig;
@@ -300,9 +303,11 @@ public void start(boolean blocking) throws Exception {
                             .getProtectionDomain().getCodeSource().getLocation().getFile();
                 }
 
-                String builtInSink = isBuiltInSink(userCodeFile);
-                if (builtInSink != null) {
-                    sinkConfig.setArchive(builtInSink);
+                if (userCodeFile != null) {
+                    String builtInSink = isBuiltInSink(userCodeFile);
+                    if (builtInSink != null) {
+                        sinkConfig.setArchive(builtInSink);
+                    }
                 }
                 parallelism = sinkConfig.getParallelism();
 
@@ -413,7 +418,7 @@ public void run() {
                             Gson gson = new GsonBuilder().setPrettyPrinting().create();
                             log.info(gson.toJson(new JsonParser().parse(json)));
                         }
-                    } catch (Exception ex) {
+                    } catch (TimeoutException | InterruptedException | ExecutionException e) {
                         log.error("Could not get status from all local instances");
                     }
                 }