Implement more Error Prone PMD rules (airbytehq#15491)

* AvoidFieldNameMatchingTypeName rule
* AvoidInstanceofChecksInCatchClause
* compareObjectsWithEquals
* DoNotTerminateVM and ConstructorCallsOverridableMethod
* EmptyIfStmt and EmptyStatementNotInLoop
* ImplicitSwitchFallThrough, InvalidLogMessageFormat, and MoreThanOneLogger

airbyte-analytics/src/main/java/io/airbyte/analytics/TrackingIdentity.java

@@ -9,6 +9,7 @@
 import java.util.Optional;
 import java.util.UUID;
 
+@SuppressWarnings("PMD.CompareObjectsWithEquals")
 public class TrackingIdentity {
 
   private final AirbyteVersion airbyteVersion;

airbyte-analytics/src/test/java/io/airbyte/analytics/SegmentTrackingClientTest.java

@@ -136,10 +136,11 @@ void testTrackWithMetadata() {
   }
 
   private static ImmutableMap<String, Object> filterTrackedAtProperty(final Map<String, ?> properties) {
-    assertTrue(properties.containsKey("tracked_at"));
+    final String trackedAtKey = "tracked_at";
+    assertTrue(properties.containsKey(trackedAtKey));
     final Builder<String, Object> builder = ImmutableMap.builder();
     properties.forEach((key, value) -> {
-      if (!"tracked_at".equals(key)) {
+      if (!trackedAtKey.equals(key)) {
         builder.put(key, value);
       }
     });

airbyte-api/src/main/java/io/airbyte/api/client/PatchedLogsApi.java

@@ -75,7 +75,7 @@ public ApiResponse<File> getLogsWithHttpInfo(final LogsRequestBody logsRequestBo
       if (memberVarResponseInterceptor != null) {
         memberVarResponseInterceptor.accept(localVarResponse);
       }
-      if (localVarResponse.statusCode() / 100 != 2) {
+      if (isErrorResponse(localVarResponse)) {
         throw new ApiException(localVarResponse.statusCode(),
             "getLogs call received non-success response",
             localVarResponse.headers(),
@@ -100,6 +100,10 @@ public ApiResponse<File> getLogsWithHttpInfo(final LogsRequestBody logsRequestBo
     }
   }
 
+  private Boolean isErrorResponse(final HttpResponse<InputStream> httpResponse) {
+    return httpResponse.statusCode() / 100 != 2;
+  }
+
   private HttpRequest.Builder getLogsRequestBuilder(final LogsRequestBody logsRequestBody) throws ApiException {
     // verify the required parameter 'logsRequestBody' is set
     if (logsRequestBody == null) {

airbyte-commons/src/main/java/io/airbyte/commons/io/FileTtlManager.java

@@ -75,7 +75,8 @@ private void reportCacheStatus() {
       }
     });
     sb.append("---\n");
-    LOGGER.info(sb.toString());
+    final String toLog = sb.toString();
+    LOGGER.info(toLog);
   }
 
 }

airbyte-commons/src/main/java/io/airbyte/commons/version/AirbyteVersion.java

@@ -10,6 +10,7 @@
 /**
  * The AirbyteVersion identifies the version of the database used internally by Airbyte services.
  */
+@SuppressWarnings("PMD.ConstructorCallsOverridableMethod")
 public class AirbyteVersion {
 
   public static final String DEV_VERSION_PREFIX = "dev";

airbyte-config/config-models/src/main/java/io/airbyte/config/EnvConfigs.java

@@ -33,7 +33,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-@SuppressWarnings({"PMD.LongVariable", "PMD.CyclomaticComplexity", "PMD.AvoidReassigningParameters"})
+@SuppressWarnings({"PMD.LongVariable", "PMD.CyclomaticComplexity", "PMD.AvoidReassigningParameters", "PMD.ConstructorCallsOverridableMethod"})
 public class EnvConfigs implements Configs {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(EnvConfigs.class);

airbyte-config/config-models/src/main/java/io/airbyte/config/helpers/StateMessageHelper.java

@@ -28,6 +28,7 @@ public static class AirbyteStateMessageListTypeReference extends TypeReference<L
    * @param state - a blob representing the state
    * @return An optional state wrapper, if there is no state an empty optional will be returned
    */
+  @SuppressWarnings("PMD.AvoidLiteralsInIfCondition")
   public static Optional<StateWrapper> getTypedState(final JsonNode state, final boolean useStreamCapableState) {
     if (state == null) {
       return Optional.empty();

airbyte-config/config-persistence/src/main/java/io/airbyte/config/persistence/StatePersistence.java

@@ -217,6 +217,7 @@ static void writeStateToDb(final DSLContext ctx,
    * @return the StateType of the records
    * @throws IllegalStateException If StateRecords have inconsistent types
    */
+  @SuppressWarnings("PMD.AvoidLiteralsInIfCondition")
   private static io.airbyte.db.instance.configs.jooq.generated.enums.StateType getStateType(
                                                                                             final UUID connectionId,
                                                                                             final List<StateRecord> records) {

airbyte-config/config-persistence/src/main/java/io/airbyte/config/persistence/split_secrets/VaultSecretPersistence.java

@@ -31,7 +31,9 @@ public Optional<String> read(final SecretCoordinate coordinate) {
       val response = vault.logical().read(pathPrefix + coordinate.getFullCoordinate());
       val restResponse = response.getRestResponse();
       val responseCode = restResponse.getStatus();
-      if (responseCode != 200) {
+      final Boolean isErrorResponse = responseCode / 100 != 2;
+
+      if (isErrorResponse) {
         log.error("Vault failed on read. Response code: " + responseCode);
         return Optional.empty();
       }

airbyte-config/config-persistence/src/test/java/io/airbyte/config/persistence/StatePersistenceTest.java

@@ -275,7 +275,7 @@ void testGlobalFullReset() throws IOException {
                         .withStreamState(null),
                     new AirbyteStreamState()
                         .withStreamDescriptor(new StreamDescriptor().withName("s1"))
-                        .withStreamState(null)))));;
+                        .withStreamState(null)))));
 
     statePersistence.updateOrCreateState(connectionId, state0);
     statePersistence.updateOrCreateState(connectionId, fullReset);

airbyte-config/init/src/main/java/io/airbyte/config/init/RemoteDefinitionsProvider.java

@@ -58,7 +58,7 @@ public void initialize() throws InterruptedException, IOException {
 
   @Override
   public StandardSourceDefinition getSourceDefinition(final UUID definitionId) throws ConfigNotFoundException {
-    StandardSourceDefinition definition = this.sourceDefinitions.get(definitionId);
+    final StandardSourceDefinition definition = this.sourceDefinitions.get(definitionId);
     if (definition == null) {
       throw new ConfigNotFoundException(SeedType.STANDARD_SOURCE_DEFINITION.name(), definitionId.toString());
     }
@@ -72,7 +72,7 @@ public List<StandardSourceDefinition> getSourceDefinitions() {
 
   @Override
   public StandardDestinationDefinition getDestinationDefinition(final UUID definitionId) throws ConfigNotFoundException {
-    StandardDestinationDefinition definition = this.destinationDefinitions.get(definitionId);
+    final StandardDestinationDefinition definition = this.destinationDefinitions.get(definitionId);
     if (definition == null) {
       throw new ConfigNotFoundException(SeedType.STANDARD_DESTINATION_DEFINITION.name(), definitionId.toString());
     }
@@ -84,15 +84,20 @@ public List<StandardDestinationDefinition> getDestinationDefinitions() {
     return new ArrayList<>(this.destinationDefinitions.values());
   }
 
-  private static CombinedConnectorCatalog getRemoteDefinitionCatalog(URI catalogUrl, Duration timeout) throws IOException, InterruptedException {
+  private static CombinedConnectorCatalog getRemoteDefinitionCatalog(final URI catalogUrl, final Duration timeout)
+      throws IOException, InterruptedException {
     final HttpRequest request = HttpRequest.newBuilder(catalogUrl).timeout(timeout).header("accept", "application/json").build();
 
     final HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
-    if (response.statusCode() >= 400) {
+    if (errorStatusCode(response)) {
       throw new IOException(
           "getRemoteDefinitionCatalog request ran into status code error: " + response.statusCode() + " with message: " + response.getClass());
     }
     return Jsons.deserialize(response.body(), CombinedConnectorCatalog.class);
   }
 
+  private static Boolean errorStatusCode(final HttpResponse<String> response) {
+    return response.statusCode() >= 400;
+  }
+
 }

airbyte-container-orchestrator/src/main/java/io/airbyte/container_orchestrator/ContainerOrchestratorApp.java

@@ -53,7 +53,7 @@
  * future this will need to independently interact with cloud storage.
  */
 @Slf4j
-@SuppressWarnings("PMD.AvoidCatchingThrowable")
+@SuppressWarnings({"PMD.AvoidCatchingThrowable", "PMD.DoNotTerminateVM"})
 public class ContainerOrchestratorApp {
 
   public static final int MAX_SECONDS_TO_WAIT_FOR_FILE_COPY = 60;

airbyte-db/db-lib/src/main/java/io/airbyte/db/DataTypeUtils.java

@@ -66,6 +66,7 @@ public static String toISO8601StringWithMicroseconds(final Instant instant) {
     return dateWithMilliseconds.substring(0, 23) + calculateMicrosecondsString(instant.getNano()) + dateWithMilliseconds.substring(23);
   }
 
+  @SuppressWarnings("PMD.AvoidLiteralsInIfCondition")
   private static String calculateMicrosecondsString(final int nano) {
     final var microSeconds = (nano / 1000) % 1000;
     final String result;

airbyte-db/db-lib/src/main/java/io/airbyte/db/IncrementalUtils.java

@@ -11,6 +11,7 @@ public class IncrementalUtils {
 
   private static final String PROPERTIES = "properties";
 
+  @SuppressWarnings("PMD.AvoidLiteralsInIfCondition")
   public static String getCursorField(final ConfiguredAirbyteStream stream) {
     if (stream.getCursorField().size() == 0) {
       throw new IllegalStateException("No cursor field specified for stream attempting to do incremental.");

airbyte-db/db-lib/src/main/java/io/airbyte/db/bigquery/BigQuerySourceOperations.java

@@ -99,7 +99,7 @@ private void setJsonField(final Field field, final FieldValue fieldValue, final
           }
         }
       } catch (final UnsupportedOperationException e) {
-        LOGGER.error("Failed to parse Object field with name: ", fieldName, e.getMessage());
+        LOGGER.error("Failed to parse Object field with name: {}, {}", fieldName, e.getMessage());
       }
     }
   }

airbyte-db/db-lib/src/main/java/io/airbyte/db/instance/configs/migrations/V0_30_22_001__Store_last_sync_state.java

@@ -106,6 +106,7 @@ static void copyData(final DSLContext ctx, final Set<StandardSyncState> standard
    * data from the job database).
    */
   @VisibleForTesting
+  @SuppressWarnings("PMD.AvoidLiteralsInIfCondition")
   static Optional<Database> getJobsDatabase(final String databaseUser, final String databasePassword, final String databaseUrl) {
     try {
       if (databaseUrl == null || "".equals(databaseUrl.trim())) {

airbyte-db/db-lib/src/main/java/io/airbyte/db/instance/jobs/migrations/V0_35_40_001__MigrateFailureReasonEnumValues.java

@@ -24,6 +24,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+@SuppressWarnings({"PMD.AvoidLiteralsInIfCondition", "PMD.CompareObjectsWithEquals"})
 public class V0_35_40_001__MigrateFailureReasonEnumValues extends BaseJavaMigration {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(V0_35_40_001__MigrateFailureReasonEnumValues.class);

airbyte-db/db-lib/src/main/java/io/airbyte/db/jdbc/JdbcUtils.java

@@ -87,6 +87,7 @@ public static Map<String, String> parseJdbcParameters(final String jdbcPropertie
     return parseJdbcParameters(jdbcPropertiesString, "&");
   }
 
+  @SuppressWarnings("PMD.AvoidLiteralsInIfCondition")
   public static Map<String, String> parseJdbcParameters(final String jdbcPropertiesString, final String delimiter) {
     final Map<String, String> parameters = new HashMap<>();
     if (!jdbcPropertiesString.isBlank()) {

airbyte-db/db-lib/src/main/java/io/airbyte/db/jdbc/streaming/BaseSizeEstimator.java

@@ -52,6 +52,7 @@ public static long getEstimatedByteSize(final Object rowData) {
    * This method ensures that the fetch size is between {@code minFetchSize} and {@code maxFetchSize},
    * inclusively.
    */
+  @SuppressWarnings("PMD.AvoidLiteralsInIfCondition")
   protected int getBoundedFetchSize() {
     if (maxRowByteSize <= 0.0) {
       return defaultFetchSize;

airbyte-db/db-lib/src/main/java/io/airbyte/db/mongodb/MongoDatabase.java

@@ -117,7 +117,7 @@ public Stream<JsonNode> read(final String collectionName, final List<String> col
           });
 
     } catch (final Exception e) {
-      LOGGER.error("Exception attempting to read data from collection: ", collectionName, e.getMessage());
+      LOGGER.error("Exception attempting to read data from collection: {}, {}", collectionName, e.getMessage());
       throw new RuntimeException(e);
     }
   }

airbyte-db/db-lib/src/main/java/io/airbyte/db/mongodb/MongoUtils.java

@@ -278,6 +278,7 @@ private static List<String> getTypes(final MongoCollection<Document> collection,
     return listOfTypes;
   }
 
+  @SuppressWarnings("PMD.AvoidLiteralsInIfCondition")
   private static BsonType getUniqueType(final List<String> types) {
     if (types.size() != 1) {
       return BsonType.STRING;

airbyte-protocol/protocol-models/src/main/java/io/airbyte/protocol/models/CatalogHelpers.java

@@ -287,7 +287,7 @@ private static boolean isOneOfField(final JsonNode schema) {
   }
 
   private static boolean isObjectWithSubFields(final Field field) {
-    return field.getType() == JsonSchemaType.OBJECT && field.getSubFields() != null && !field.getSubFields().isEmpty();
+    return field.getType().equals(JsonSchemaType.OBJECT) && field.getSubFields() != null && !field.getSubFields().isEmpty();
   }
 
   public static StreamDescriptor extractStreamDescriptor(final AirbyteStream airbyteStream) {

airbyte-protocol/protocol-models/src/main/java/io/airbyte/protocol/models/CommonField.java

@@ -20,7 +20,7 @@ public CommonField(final String name, final T type) {
     this.properties = null;
   }
 
-  public CommonField(final String name, final T type, List<CommonField<T>> properties) {
+  public CommonField(final String name, final T type, final List<CommonField<T>> properties) {
     this.name = name;
     this.type = type;
     this.properties = properties;
@@ -45,7 +45,7 @@ public boolean equals(final Object o) {
 
     final CommonField<T> field = (CommonField<T>) o;
     return name.equals(field.name) &&
-        type == field.type && Objects.equals(properties, field.properties);
+        type.equals(field.type) && Objects.equals(properties, field.properties);
   }
 
   @Override

airbyte-scheduler/scheduler-models/src/main/java/io/airbyte/scheduler/models/Job.java

@@ -91,6 +91,7 @@ public long getUpdatedAtInSecond() {
     return updatedAtInSecond;
   }
 
+  @SuppressWarnings("PMD.AvoidLiteralsInIfCondition")
   public Optional<Attempt> getSuccessfulAttempt() {
     final List<Attempt> successfulAttempts = getAttempts()
         .stream()

airbyte-scheduler/scheduler-persistence/src/main/java/io/airbyte/scheduler/persistence/DefaultJobPersistence.java

@@ -643,7 +643,7 @@ public void setDeployment(final UUID deployment) throws IOException {
         .orElse(deployment); // if no record was returned that means that the new deployment id was used.
 
     if (!deployment.equals(committedDeploymentId)) {
-      LOGGER.warn("Attempted to set a deployment id %s, but deployment id %s already set. Retained original value.");
+      LOGGER.warn("Attempted to set a deployment id {}, but deployment id {} already set. Retained original value.", deployment, deployment);
     }
   }
 

airbyte-scheduler/scheduler-persistence/src/main/java/io/airbyte/scheduler/persistence/JobCleaner.java

@@ -84,6 +84,7 @@ private void deleteOldFiles() throws IOException {
         });
   }
 
+  @SuppressWarnings("PMD.AvoidLiteralsInIfCondition")
   private void deleteOnSize() throws IOException {
     final Set<String> nonTerminalJobIds = new HashSet<>();
     final Sets.SetView<JobStatus> nonTerminalStatuses = Sets.difference(Set.of(JobStatus.values()), JobStatus.TERMINAL_STATUSES);

airbyte-scheduler/scheduler-persistence/src/main/java/io/airbyte/scheduler/persistence/job_error_reporter/SentryExceptionHelper.java

@@ -20,6 +20,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+@SuppressWarnings("PMD.AvoidLiteralsInIfCondition")
 public class SentryExceptionHelper {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(SentryExceptionHelper.class);

airbyte-server/src/main/java/io/airbyte/server/ServerApp.java

@@ -100,6 +100,7 @@ public ServerApp(final AirbyteVersion airbyteVersion,
   }
 
   @Override
+  @SuppressWarnings("PMD.InvalidLogMessageFormat")
   public void start() throws Exception {
     final Server server = new Server(PORT);
 
@@ -138,7 +139,7 @@ public void start() throws Exception {
     Runtime.getRuntime().addShutdownHook(new Thread(() -> {
       try {
         server.stop();
-      } catch (Exception ex) {
+      } catch (final Exception ex) {
         // silently fail at this stage because server is terminating.
         LOGGER.warn("exception: " + ex);
       }

airbyte-server/src/main/java/io/airbyte/server/handlers/ConnectionsHandler.java

@@ -317,12 +317,8 @@ public Set<StreamDescriptor> getConfigurationDiff(final AirbyteCatalog oldCatalo
     newStreams.forEach(((streamDescriptor, airbyteStreamConfiguration) -> {
       final AirbyteStreamConfiguration oldConfig = oldStreams.get(streamDescriptor);
 
-      if (oldConfig == null) {
-        // The stream is a new one, the config has not change and it needs to be in the schema change list.
-      } else {
-        if (haveConfigChange(oldConfig, airbyteStreamConfiguration)) {
-          streamWithDifferentConf.add(streamDescriptor);
-        }
+      if (oldConfig != null && haveConfigChange(oldConfig, airbyteStreamConfiguration)) {
+        streamWithDifferentConf.add(streamDescriptor);
       }
     }));