Native image compatible Kafka connector #2346

common/common/src/main/java/io/helidon/common/FeatureCatalog.java

@@ -361,7 +361,7 @@ final class FeatureCatalog {
                     .description("Reactive messaging connector for Kafka")
                     .path("Messaging", "Kafka")
                     .experimental(true)
-                    .nativeSupported(false));
+                    .nativeSupported(true));
         add("io.helidon.messaging.connectors.jms",
             FeatureDescriptor.builder()
                     .name("JMS Connector")

messaging/kafka/pom.xml

@@ -64,6 +64,11 @@
             <scope>provided</scope>
             <optional>true</optional>
         </dependency>
+        <dependency>
+            <groupId>org.graalvm.nativeimage</groupId>
+            <artifactId>svm</artifactId>
+            <scope>provided</scope>
+        </dependency>
         <dependency>
             <groupId>org.apache.kafka</groupId>
             <artifactId>kafka-clients</artifactId>

messaging/kafka/src/main/java/io/helidon/messaging/connectors/kafka/AppInfoParserSubstitution.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright (c) 2020 Oracle and/or its affiliates.
+ *
+ * Licensed 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 io.helidon.messaging.connectors.kafka;
+
+import com.oracle.svm.core.annotate.Substitute;
+import com.oracle.svm.core.annotate.TargetClass;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.utils.AppInfoParser;
+
+/**
+ * JMX not supported in native-image.
+ */
+@TargetClass(AppInfoParser.class)
+@SuppressWarnings("checkstyle:HideUtilityClassConstructor")
+final class AppInfoParserSubstitution {
+
+    @Substitute
+    public static void registerAppInfo(String p, String i, Metrics m, long n) {
+    }
+
+    @Substitute
+    public static void unregisterAppInfo(String p, String i, Metrics m) {
+    }
+}

messaging/kafka/src/main/java/io/helidon/messaging/connectors/kafka/CompressionTypeSubstitution.java

@@ -0,0 +1,136 @@
+/*
+ * Copyright (c) 2020 Oracle and/or its affiliates.
+ *
+ * Licensed 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 io.helidon.messaging.connectors.kafka;
+
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.lang.reflect.Constructor;
+import java.nio.ByteBuffer;
+
+import io.helidon.common.LazyValue;
+
+import com.oracle.svm.core.annotate.Substitute;
+import com.oracle.svm.core.annotate.TargetClass;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.record.BufferSupplier;
+import org.apache.kafka.common.utils.ByteBufferInputStream;
+import org.apache.kafka.common.utils.ByteBufferOutputStream;
+
+/**
+ * Helper for creating ZSTD or SNAPPY compression stream wrappers without method handles.
+ */
+@SuppressWarnings("checkstyle:OuterTypeFilename")
+final class CompressionTypeHelper {
+
+    private CompressionTypeHelper() {
+    }
+
+    static final LazyValue<Constructor<?>> LAZY_INPUT_ZSTD =
+            LazyValue.create(() -> findConstructor("com.github.luben.zstd.ZstdInputStream", InputStream.class));
+    static final LazyValue<Constructor<?>> LAZY_OUTPUT_ZSTD =
+            LazyValue.create(() -> findConstructor("com.github.luben.zstd.ZstdOutputStream", OutputStream.class));
+    static final LazyValue<Constructor<?>> LAZY_INPUT_SNAPPY =
+            LazyValue.create(() -> findConstructor("org.xerial.snappy.SnappyInputStream", InputStream.class));
+    static final LazyValue<Constructor<?>> LAZY_OUTPUT_SNAPPY =
+            LazyValue.create(() -> findConstructor("org.xerial.snappy.SnappyOutputStream", OutputStream.class));
+
+    static OutputStream snappyOutputStream(OutputStream orig) {
+        try {
+            return (OutputStream) LAZY_OUTPUT_SNAPPY.get().newInstance(orig);
+        } catch (KafkaException e) {
+            throw e;
+        } catch (Exception e) {
+            throw new KafkaException(e);
+        }
+    }
+
+    static InputStream snappyInputStream(ByteBuffer orig) {
+        try {
+            return (InputStream) LAZY_INPUT_SNAPPY.get().newInstance(new ByteBufferInputStream(orig));
+        } catch (KafkaException e) {
+            throw e;
+        } catch (Exception e) {
+            throw new KafkaException(e);
+        }
+    }
+
+    static OutputStream zstdOutputStream(OutputStream orig) {
+        try {
+            return (OutputStream) LAZY_OUTPUT_ZSTD.get().newInstance(orig);
+        } catch (KafkaException e) {
+            throw e;
+        } catch (Exception e) {
+            throw new KafkaException(e);
+        }
+    }
+
+    static InputStream zstdInputStream(ByteBuffer orig) {
+        try {
+            return (InputStream) LAZY_INPUT_ZSTD.get().newInstance(new ByteBufferInputStream(orig));
+        } catch (KafkaException e) {
+            throw e;
+        } catch (Exception e) {
+            throw new KafkaException(e);
+        }
+    }
+
+    static Constructor<?> findConstructor(String className, Class<?>... paramTypes) {
+        try {
+            return Class.forName(className)
+                    .getDeclaredConstructor(paramTypes);
+        } catch (NoSuchMethodException | ClassNotFoundException e) {
+            throw new KafkaException(e);
+        }
+    }
+}
+
+/**
+ * Substitution for {@link org.apache.kafka.common.record.CompressionType#SNAPPY CompressionType.SNAPPY}.
+ */
+@TargetClass(className = "org.apache.kafka.common.record.CompressionType$3")
+@SuppressWarnings("checkstyle:OneTopLevelClass")
+final class SnappySubstitution {
+
+    @Substitute
+    public OutputStream wrapForOutput(ByteBufferOutputStream buffer, byte messageVersion) {
+        return CompressionTypeHelper.snappyOutputStream(buffer);
+    }
+
+    @Substitute
+    public InputStream wrapForInput(ByteBuffer buffer, byte messageVersion, BufferSupplier decompressionBufferSupplier) {
+        return CompressionTypeHelper.snappyInputStream(buffer);
+    }
+}
+
+/**
+ * Substitution for {@link org.apache.kafka.common.record.CompressionType#ZSTD CompressionType.ZSTD}.
+ */
+@TargetClass(className = "org.apache.kafka.common.record.CompressionType$5")
+@SuppressWarnings("checkstyle:OneTopLevelClass")
+final class ZstdSubstitution {
+
+    @Substitute
+    public OutputStream wrapForOutput(ByteBufferOutputStream buffer, byte messageVersion) {
+        return CompressionTypeHelper.zstdOutputStream(buffer);
+    }
+
+    @Substitute
+    public InputStream wrapForInput(ByteBuffer buffer, byte messageVersion, BufferSupplier decompressionBufferSupplier) {
+        return CompressionTypeHelper.zstdInputStream(buffer);
+    }
+}

messaging/kafka/src/main/java/io/helidon/messaging/connectors/kafka/Crc32CSubstitution.java

@@ -0,0 +1,58 @@
+/*
+ * Copyright (c) 2020 Oracle and/or its affiliates.
+ *
+ * Licensed 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 io.helidon.messaging.connectors.kafka;
+
+import java.nio.ByteBuffer;
+import java.util.zip.Checksum;
+
+import com.oracle.svm.core.annotate.Substitute;
+import com.oracle.svm.core.annotate.TargetClass;
+import org.apache.kafka.common.utils.Checksums;
+
+
+/**
+ * Method handles are not supported by native-image,
+ * invoke {@link java.util.zip.CRC32C CRC32C} directly.
+ *
+ * Helidon runs only on Java 11 and newer, {@link java.util.zip.CRC32C CRC32C}
+ * doesn't have to be instantiated by method handles.
+ */
+@TargetClass(org.apache.kafka.common.utils.Crc32C.class)
+@Substitute
+@SuppressWarnings("checkstyle:HideUtilityClassConstructor")
+final class Crc32CSubstitution {
+
+    @Substitute
+    public static long compute(byte[] bytes, int offset, int size) {
+        Checksum crc = create();
+        crc.update(bytes, offset, size);
+        return crc.getValue();
+    }
+
+    @Substitute
+    public static long compute(ByteBuffer buffer, int offset, int size) {
+        Checksum crc = create();
+        Checksums.update(crc, buffer, offset, size);
+        return crc.getValue();
+    }
+
+    @Substitute
+    public static Checksum create() {
+        return new java.util.zip.CRC32C();
+    }
+}

messaging/kafka/src/main/java/io/helidon/messaging/connectors/kafka/JmxReporterSubstitution.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright (c) 2020 Oracle and/or its affiliates.
+ *
+ * Licensed 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 io.helidon.messaging.connectors.kafka;
+
+import com.oracle.svm.core.annotate.Substitute;
+import com.oracle.svm.core.annotate.TargetClass;
+import org.apache.kafka.common.metrics.KafkaMetric;
+
+/**
+ * JMX not supported in native-image.
+ */
+@TargetClass(org.apache.kafka.common.metrics.JmxReporter.class)
+final class JmxReporterSubstitution {
+
+    @Substitute
+    private Object addAttribute(KafkaMetric metric) {
+        return null;
+    }
+
+    @Substitute
+    public void metricChange(KafkaMetric metric) {
+    }
+
+}
+

messaging/kafka/src/main/java/module-info.java

@@ -30,6 +30,7 @@
     requires io.helidon.common.configurable;
     requires io.helidon.messaging;
     requires microprofile.config.api;
+    requires svm;
 
     exports io.helidon.messaging.connectors.kafka;
 }

messaging/kafka/src/main/resources/META-INF/helidon/native-image/reflection-config.json

@@ -0,0 +1,14 @@
+{
+    "annotated":[
+    ],
+    "class-hierarchy": [
+        "org.apache.kafka.common.serialization.Serializer",
+        "org.apache.kafka.common.serialization.Deserializer",
+        "org.apache.kafka.clients.consumer.ConsumerPartitionAssignor",
+        "org.apache.kafka.clients.producer.Partitioner"
+    ],
+    "classes": [
+    ],
+    "exclude": [
+    ]
+}

messaging/kafka/src/main/resources/META-INF/native-image/io.helidon.messaging.connectors.kafka/native-image.properties

@@ -0,0 +1,25 @@
+#
+# Copyright (c) 2019, 2020 Oracle and/or its affiliates.
+#
+# Licensed 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.
+#
+
+Args=--initialize-at-build-time=org.slf4j.impl.SimpleLogger \
+  --initialize-at-build-time=org.slf4j.impl.StaticLoggerBinder \
+  --initialize-at-build-time=org.slf4j.LoggerFactory \
+  --initialize-at-build-time=org.xerial.snappy.SnappyInputStream \
+  --initialize-at-build-time=org.xerial.snappy.SnappyOutputStream \
+  --initialize-at-build-time=com.github.luben.zstd.ZstdInputStream \
+  --initialize-at-build-time=com.github.luben.zstd.ZstdOutputStream \
+  --initialize-at-build-time=org.apache.kafka \
+  --report-unsupported-elements-at-runtime

messaging/kafka/src/main/resources/META-INF/native-image/reflect-config.json

@@ -0,0 +1,2 @@
+[
+]