Enable CheckStyle Plugin in Pulsar Io (apache#13413)

buildtools/src/main/resources/pulsar/suppressions.xml

@@ -46,4 +46,41 @@
 
     <suppress checks=".*" files="MLDataFormats.java"/>
     <suppress checks=".*" files="BitSetRecyclable.java"/>
+    <!-- pulsar io -->
+    <suppress checks="RegexpSinglelineJava" files="KafkaConnectData.java"/>
+    <suppress checks="LineLength" files="KafkaConnectData.java"/>
+    <suppress checks="RegexpSinglelineJava" files="KafkaConnectSink.java"/>
+    <!-- pulsar io es -->
+    <suppress checks="LineLength" files="ElasticSearchConfig.java"/>
+    <suppress checks="LineLength" files="ElasticSearchClient.java"/>
+    <!-- pulsar io flume -->
+    <suppress checks="RegexpSingleline" files="Application.java"/>
+    <suppress checks="RegexpSinglelineJava" files="AbstractConfigurationProvider.java"/>
+    <suppress checks="RegexpSinglelineJava" files="AbstractZooKeeperConfigurationProvider.java"/>
+    <suppress checks="RegexpSinglelineJava" files="PollingPropertiesFileConfigurationProvider.java"/>
+    <suppress checks="RegexpSinglelineJava" files="EnvVarResolverProperties.java"/>
+    <!-- pulsar io hbase -->
+    <suppress checks="RegexpSinglelineJava" files="HbaseAbstractSink.java"/>
+    <suppress checks="RegexpSinglelineJava" files="HbaseAbstractConfig.java"/>
+    <suppress checks="RegexpSinglelineJava" files="HbaseSinkConfig.java"/>
+    <!-- pulsar io kafka -->
+    <suppress checks="FallThrough" files="KafkaConnectData.java"/>
+    <!-- pulsar io influxdb -->
+    <suppress checks="RegexpSinglelineJava" files="InfluxDBSinkConfig.java"/>
+    <!-- pulsar io kinesis -->
+    <suppress checks="LineLength" files="KinesisSinkConfig.java"/>
+    <suppress checks="LineLength" files="KinesisSink.java"/>
+    <suppress checks=".*" files=".+[\\/]org[\\/]apache[\\/]pulsar[\\/]io[\\/]kinesis[\\/]fbs[\\/].+.java"/>
+    <!-- pulsar io mongo -->
+    <suppress checks="RegexpSinglelineJava" files="MongoSink.java"/>
+    <suppress checks="RegexpSinglelineJava" files="MongoConfig.java"/>
+    <!-- pulsar io rabbitmq -->
+    <suppress checks="RegexpSinglelineJava" files="RabbitMQAbstractConfig.java"/>
+    <suppress checks="RegexpSinglelineJava" files="RabbitMQSourceConfig.java"/>
+    <suppress checks="RegexpSinglelineJava" files="RabbitMQSinkConfig.java"/>
+    <!-- pulsar io reids -->
+    <suppress checks="RegexpSinglelineJava" files="RedisAbstractConfig.java"/>
+    <suppress checks="RegexpSinglelineJava" files="RedisSinkConfig.java"/>
+    <!-- pulsar io solr -->
+    <suppress checks="RegexpSinglelineJava" files="SolrSinkConfig.java"/>
 </suppressions>

pulsar-io/aerospike/src/main/java/org/apache/pulsar/io/aerospike/AerospikeAbstractSink.java

@@ -31,11 +31,9 @@
 import com.aerospike.client.listener.WriteListener;
 import com.aerospike.client.policy.ClientPolicy;
 import com.aerospike.client.policy.WritePolicy;
-
 import java.util.Map;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingDeque;
-
 import org.apache.pulsar.functions.api.Record;
 import org.apache.pulsar.io.core.KeyValue;
 import org.apache.pulsar.io.core.Sink;
@@ -44,7 +42,7 @@
 import org.slf4j.LoggerFactory;
 
 /**
- * A Simple abstract class for Aerospike sink
+ * A Simple abstract class for Aerospike sink.
  * Users need to implement extractKeyValue function to use this sink
  */
 public abstract class AerospikeAbstractSink<K, V> implements Sink<byte[]> {
@@ -95,7 +93,8 @@ public void close() throws Exception {
     @Override
     public void write(Record<byte[]> record) {
         KeyValue<K, V> keyValue = extractKeyValue(record);
-        Key key = new Key(aerospikeSinkConfig.getKeyspace(), aerospikeSinkConfig.getKeySet(), keyValue.getKey().toString());
+        Key key = new Key(aerospikeSinkConfig.getKeyspace(), aerospikeSinkConfig.getKeySet(),
+                keyValue.getKey().toString());
         Bin bin = new Bin(aerospikeSinkConfig.getColumnName(), Value.getAsBlob(keyValue.getValue()));
         AWriteListener listener = null;
         try {
@@ -148,7 +147,7 @@ public void onSuccess(Key key) {
             try {
                 queue.put(this);
             } catch (InterruptedException ex) {
-                throw new RuntimeException("Interrupted while being added to the queue" ,ex);
+                throw new RuntimeException("Interrupted while being added to the queue", ex);
             }
         }
 

pulsar-io/aerospike/src/main/java/org/apache/pulsar/io/aerospike/AerospikeSinkConfig.java

@@ -21,13 +21,12 @@
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
-import lombok.Data;
-import lombok.experimental.Accessors;
-
 import java.io.File;
 import java.io.IOException;
 import java.io.Serializable;
 import java.util.Map;
+import lombok.Data;
+import lombok.experimental.Accessors;
 
 @Data
 @Accessors(chain = true)

pulsar-io/aerospike/src/main/java/org/apache/pulsar/io/aerospike/AerospikeStringSink.java

@@ -19,9 +19,9 @@
 
 package org.apache.pulsar.io.aerospike;
 
+import java.nio.charset.StandardCharsets;
 import org.apache.pulsar.functions.api.Record;
 import org.apache.pulsar.io.core.KeyValue;
-import java.nio.charset.StandardCharsets;
 
 /**
  * Aerospike sink that treats incoming messages on the input topic as Strings

pulsar-io/aerospike/src/main/java/org/apache/pulsar/io/aerospike/package-info.java

@@ -0,0 +1,19 @@
+/**
+ * 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.pulsar.io.aerospike;

pulsar-io/aws/src/main/java/org/apache/pulsar/io/aws/AbstractAwsConnector.java

@@ -18,22 +18,19 @@
  */
 package org.apache.pulsar.io.aws;
 
-import java.io.IOException;
-import java.lang.reflect.Constructor;
-import java.util.Map;
-
 import com.amazonaws.auth.AWSCredentials;
 import com.amazonaws.auth.AWSCredentialsProvider;
 import com.amazonaws.auth.BasicAWSCredentials;
 import com.google.gson.Gson;
 import com.google.gson.reflect.TypeToken;
-
+import java.io.IOException;
+import java.lang.reflect.Constructor;
+import java.util.Map;
 import lombok.extern.slf4j.Slf4j;
 import software.amazon.awssdk.utils.StringUtils;
 
 @Slf4j
 public abstract class AbstractAwsConnector {
-
     public static final String ACCESS_KEY_NAME = "accessKey";
     public static final String SECRET_KEY_NAME = "secretKey";
 
@@ -69,9 +66,9 @@ public static AwsCredentialProviderPlugin createCredentialProviderWithPlugin(Str
                     String.format("invalid authplugin name %s , failed to init %s", pluginFQClassName, e.getMessage()));
         }
     }
-    
+
     /**
-     * It creates a default credential provider which takes accessKey and secretKey form configuration and creates
+     * It creates a default credential provider which takes accessKey and secretKey form configuration and creates.
      * {@link AWSCredentials}
      *
      * @param awsCredentialPluginParam
@@ -84,11 +81,8 @@ public AwsCredentialProviderPlugin defaultCredentialProvider(String awsCredentia
         String accessKey = credentialMap.get(ACCESS_KEY_NAME);
         String secretKey = credentialMap.get(SECRET_KEY_NAME);
         if (!(StringUtils.isNotBlank(accessKey) && StringUtils.isNotBlank(secretKey))) {
-            throw new IllegalArgumentException(
-                    String.format(
-                            "Default %s and %s must be present into json-map if AwsCredentialProviderPlugin not provided",
-                            ACCESS_KEY_NAME, SECRET_KEY_NAME)
-            );
+            throw new IllegalArgumentException(String.format("Default %s and %s must be present into json-map "
+                    + "if AwsCredentialProviderPlugin not provided", ACCESS_KEY_NAME, SECRET_KEY_NAME));
         }
         return new AwsCredentialProviderPlugin() {
             @Override
@@ -108,7 +102,7 @@ public void close() throws IOException {
             }
         };
     }
-    
+
     private AWSCredentialsProvider defaultCredentialProvider(String accessKey, String secretKey) {
         return new AWSCredentialsProvider() {
             @Override

pulsar-io/aws/src/main/java/org/apache/pulsar/io/aws/AwsCredentialProviderPlugin.java

@@ -19,12 +19,11 @@
 
 package org.apache.pulsar.io.aws;
 
-import java.io.Closeable;
-
 import com.amazonaws.auth.AWSCredentials;
 import com.amazonaws.auth.AWSCredentialsProvider;
 import com.amazonaws.auth.AWSSessionCredentials;
 import com.amazonaws.auth.BasicSessionCredentials;
+import java.io.Closeable;
 
 /**
  * Kinesis source/sink calls credential-provider while refreshing aws accessKey and secreKey. So, implementation
@@ -35,7 +34,6 @@ public interface AwsCredentialProviderPlugin extends Closeable {
 
     /**
      * accepts aws-account related param and initialize credential provider.
-     * 
      * @param param
      */
     void init(String param);
@@ -44,7 +42,6 @@ public interface AwsCredentialProviderPlugin extends Closeable {
      * Returned {@link AWSCredentialsProvider} can give {@link AWSCredentials} in case credential belongs to IAM user or
      * it can return {@link BasicSessionCredentials} if user wants to generate temporary credential for a given IAM
      * role.
-     * 
      * @return
      */
     AWSCredentialsProvider getCredentialProvider();

pulsar-io/aws/src/main/java/org/apache/pulsar/io/aws/AwsDefaultProviderChainPlugin.java

@@ -20,9 +20,8 @@
 
 import com.amazonaws.auth.AWSCredentialsProvider;
 import com.amazonaws.auth.DefaultAWSCredentialsProviderChain;
-import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
-
 import java.io.IOException;
+import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
 
 public class AwsDefaultProviderChainPlugin implements AwsCredentialProviderPlugin {
     @Override

pulsar-io/aws/src/main/java/org/apache/pulsar/io/aws/STSAssumeRoleProviderPlugin.java

@@ -22,11 +22,10 @@
 import com.amazonaws.auth.STSAssumeRoleSessionCredentialsProvider;
 import com.google.gson.Gson;
 import com.google.gson.reflect.TypeToken;
-import software.amazon.awssdk.services.sts.StsClient;
-import software.amazon.awssdk.services.sts.auth.StsAssumeRoleCredentialsProvider;
-
 import java.io.IOException;
 import java.util.Map;
+import software.amazon.awssdk.services.sts.StsClient;
+import software.amazon.awssdk.services.sts.auth.StsAssumeRoleCredentialsProvider;
 
 public class STSAssumeRoleProviderPlugin implements AwsCredentialProviderPlugin {
     public static final String ASSUME_ROLE_ARN = "roleArn";

pulsar-io/aws/src/main/java/org/apache/pulsar/io/aws/package-info.java

@@ -0,0 +1,19 @@
+/**
+ * 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.pulsar.io.aws;

pulsar-io/batch-data-generator/src/main/java/org/apache/pulsar/io/batchdatagenerator/BatchDataGeneratorPrintSink.java

@@ -18,13 +18,12 @@
  */
 package org.apache.pulsar.io.batchdatagenerator;
 
+import java.util.Map;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.functions.api.Record;
 import org.apache.pulsar.io.core.Sink;
 import org.apache.pulsar.io.core.SinkContext;
 
-import java.util.Map;
-
 @Slf4j
 public class BatchDataGeneratorPrintSink implements Sink<Person> {
 

pulsar-io/batch-data-generator/src/main/java/org/apache/pulsar/io/batchdatagenerator/BatchDataGeneratorPushSource.java

@@ -20,16 +20,15 @@
 package org.apache.pulsar.io.batchdatagenerator;
 
 import io.codearte.jfairy.Fairy;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.pulsar.functions.api.Record;
-import org.apache.pulsar.io.core.BatchPushSource;
-import org.apache.pulsar.io.core.SourceContext;
-
 import java.nio.charset.StandardCharsets;
 import java.util.Map;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.function.Consumer;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.functions.api.Record;
+import org.apache.pulsar.io.core.BatchPushSource;
+import org.apache.pulsar.io.core.SourceContext;
 
 @Slf4j
 public class BatchDataGeneratorPushSource extends BatchPushSource<Person> implements Runnable {

pulsar-io/batch-data-generator/src/main/java/org/apache/pulsar/io/batchdatagenerator/BatchDataGeneratorSource.java

@@ -19,15 +19,14 @@
 package org.apache.pulsar.io.batchdatagenerator;
 
 import io.codearte.jfairy.Fairy;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.pulsar.functions.api.Record;
-import org.apache.pulsar.io.core.BatchSource;
-import org.apache.pulsar.io.core.SourceContext;
-
 import java.nio.charset.StandardCharsets;
 import java.util.Map;
 import java.util.Optional;
 import java.util.function.Consumer;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.functions.api.Record;
+import org.apache.pulsar.io.core.BatchSource;
+import org.apache.pulsar.io.core.SourceContext;
 
 @Slf4j
 public class BatchDataGeneratorSource implements BatchSource<Person> {

pulsar-io/batch-data-generator/src/main/java/org/apache/pulsar/io/batchdatagenerator/package-info.java

@@ -0,0 +1,19 @@
+/**
+ * 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.pulsar.io.batchdatagenerator;

pulsar-io/batch-discovery-triggerers/src/main/java/org/apache/pulsar/io/batchdiscovery/CronTriggerer.java

@@ -18,16 +18,15 @@
  */
 package org.apache.pulsar.io.batchdiscovery;
 
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.Consumer;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.io.core.BatchSourceTriggerer;
 import org.apache.pulsar.io.core.SourceContext;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
 import org.springframework.scheduling.support.CronTrigger;
 
-import java.util.Map;
-import java.util.Objects;
-import java.util.function.Consumer;
-
 /**
  * This is an implementation of BatchSourceTriggerer that triggers based on a cron expression.
  * BatchSource developers using this should pass the json string of a map that contains
@@ -49,7 +48,7 @@ public void init(Map<String, Object> config, SourceContext sourceContext) {
     }
     scheduler = new ThreadPoolTaskScheduler();
     scheduler.setThreadNamePrefix(String.format("%s/%s/%s-cron-triggerer-",
-      sourceContext.getTenant(), sourceContext.getNamespace(), sourceContext.getSourceName()));
+            sourceContext.getTenant(), sourceContext.getNamespace(), sourceContext.getSourceName()));
 
     log.info("Initialized CronTrigger with expression: {}", cronExpression);
   }

pulsar-io/batch-discovery-triggerers/src/main/java/org/apache/pulsar/io/batchdiscovery/ImmediateTriggerer.java

@@ -20,17 +20,16 @@
 
 import java.util.Map;
 import java.util.function.Consumer;
-
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.io.core.BatchSourceTriggerer;
 import org.apache.pulsar.io.core.SourceContext;
 
 @Slf4j
 public class ImmediateTriggerer implements BatchSourceTriggerer {
-	
+
   @Override
   public void init(Map<String, Object> map, SourceContext sourceContext) throws Exception {
-    log.info("Initialized ImmediateTrigger at: {}",  System.currentTimeMillis());
+    log.info("Initialized ImmediateTrigger at: {}", System.currentTimeMillis());
   }
 
   @Override