fix(filesystems): make compression codec more robust to encoding

Author: fhussonnois

connect-file-pulse-filesystems/filepulse-local-fs/src/main/java/io/streamthoughts/kafka/connect/filepulse/fs/codec/GZipCodec.java

@@ -26,15 +26,17 @@
 import java.io.FileInputStream;
 import java.io.IOException;
 import java.nio.file.Files;
+import java.util.Arrays;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.Set;
 import java.util.zip.GZIPInputStream;
 
 public class GZipCodec implements CodecHandler {
 
-    private static Logger LOG = LoggerFactory.getLogger(GZipCodec.class);
+    private static final Logger LOG = LoggerFactory.getLogger(GZipCodec.class);
 
     private static final Set<String> MIME_TYPES = new HashSet<>();
 
@@ -57,8 +59,11 @@ public void configure(final Map<String, ?> configs) {
     public boolean canRead(final File file) {
         Objects.requireNonNull(file, "file can't be null");
         try {
-            final String type = Files.probeContentType(file.toPath());
-            return MIME_TYPES.contains(type);
+            final Optional<String> contentType = Optional.ofNullable(Files.probeContentType(file.toPath()));
+            return contentType.map(s -> Arrays.stream(s.split(";"))
+                    .map(String::trim)
+                    .anyMatch(MIME_TYPES::contains))
+                    .orElse(false);
         } catch (IOException e) {
             LOG.warn("Unexpected error occurred while proving content-type for file : {}", file.getAbsolutePath());
             return false;

connect-file-pulse-filesystems/filepulse-local-fs/src/main/java/io/streamthoughts/kafka/connect/filepulse/fs/codec/TarballCodec.java

@@ -37,7 +37,7 @@
 
 public class TarballCodec implements CodecHandler {
 
-    private static Logger LOG = LoggerFactory.getLogger(GZipCodec.class);
+    private static final Logger LOG = LoggerFactory.getLogger(GZipCodec.class);
 
     private static final String TAR_GZ_SUFFIX = ".tar.gz";
 

connect-file-pulse-filesystems/filepulse-local-fs/src/main/java/io/streamthoughts/kafka/connect/filepulse/fs/codec/ZipCodec.java

@@ -27,9 +27,11 @@
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Paths;
+import java.util.Arrays;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.Set;
 import java.util.zip.ZipEntry;
 import java.util.zip.ZipInputStream;
@@ -60,8 +62,11 @@ public void configure(final Map<String, ?> configs) {
     public boolean canRead(final File file) {
         Objects.requireNonNull(file, "file can't be null");
         try {
-            final String type = Files.probeContentType(file.toPath());
-            return MIME_TYPES.contains(type);
+            final Optional<String> contentType = Optional.ofNullable(Files.probeContentType(file.toPath()));
+            return contentType.map(s -> Arrays.stream(s.split(";"))
+                    .map(String::trim)
+                    .anyMatch(MIME_TYPES::contains))
+                    .orElse(false);
         } catch (IOException e) {
             LOG.warn("Unexpected error occurred while proving content-type for file : {}", file.getAbsolutePath());
             return false;