diff --git a/extensions/aws2-kinesis/deployment/pom.xml b/extensions/aws2-kinesis/deployment/pom.xml
index 194dd861a52f..fe5353eab76f 100644
--- a/extensions/aws2-kinesis/deployment/pom.xml
+++ b/extensions/aws2-kinesis/deployment/pom.xml
@@ -42,11 +42,14 @@
org.apache.camel.quarkus
camel-quarkus-aws2-kinesis
-
io.quarkus
quarkus-netty-deployment
+
+ org.eclipse.transformer
+ org.eclipse.transformer
+
diff --git a/extensions/aws2-kinesis/deployment/src/main/java/org/apache/camel/quarkus/component/aws2/kinesis/deployment/JakartaEnablement.java b/extensions/aws2-kinesis/deployment/src/main/java/org/apache/camel/quarkus/component/aws2/kinesis/deployment/JakartaEnablement.java
new file mode 100644
index 000000000000..232a6fbe2d0c
--- /dev/null
+++ b/extensions/aws2-kinesis/deployment/src/main/java/org/apache/camel/quarkus/component/aws2/kinesis/deployment/JakartaEnablement.java
@@ -0,0 +1,86 @@
+/*
+ * 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.camel.quarkus.component.aws2.kinesis.deployment;
+
+import java.nio.ByteBuffer;
+import java.util.Collections;
+import java.util.Map;
+
+import io.quarkus.bootstrap.classloading.QuarkusClassLoader;
+import io.quarkus.deployment.annotations.BuildProducer;
+import io.quarkus.deployment.annotations.BuildStep;
+import io.quarkus.deployment.builditem.BytecodeTransformerBuildItem;
+import org.eclipse.transformer.action.ActionContext;
+import org.eclipse.transformer.action.ByteData;
+import org.eclipse.transformer.action.impl.ActionContextImpl;
+import org.eclipse.transformer.action.impl.ByteDataImpl;
+import org.eclipse.transformer.action.impl.ClassActionImpl;
+import org.eclipse.transformer.action.impl.SelectionRuleImpl;
+import org.eclipse.transformer.action.impl.SignatureRuleImpl;
+import org.eclipse.transformer.util.FileUtils;
+import org.objectweb.asm.ClassReader;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Transforms usage of javax.xml.bind in software.amazon.kinesis.retrieval.AggregatorUtil. TODO: Remove this:
+ * https://github.com/apache/camel-quarkus/issues/6238
+ */
+public class JakartaEnablement {
+ @BuildStep
+ void transformToJakarta(BuildProducer transformers) {
+ if (QuarkusClassLoader.isClassPresentAtRuntime("jakarta.xml.bind.DatatypeConverter")) {
+ JakartaTransformer tr = new JakartaTransformer();
+ final BytecodeTransformerBuildItem item = new BytecodeTransformerBuildItem.Builder()
+ .setCacheable(true)
+ .setContinueOnFailure(false)
+ .setClassToTransform("software.amazon.kinesis.retrieval.AggregatorUtil")
+ .setClassReaderOptions(ClassReader.SKIP_DEBUG)
+ .setInputTransformer(tr::transform)
+ .build();
+ transformers.produce(item);
+ }
+ }
+
+ private static class JakartaTransformer {
+
+ private final Logger logger;
+ private final ActionContext ctx;
+ // We need to prevent the Eclipse Transformer to adjust the "javax" packages.
+ // Thus why we split the strings.
+ private static final Map renames = Map.of("javax" + ".xml.bind", "jakarta.xml.bind");
+
+ JakartaTransformer() {
+ logger = LoggerFactory.getLogger("JakartaTransformer");
+ //N.B. we enable only this single transformation of package renames, not the full set of capabilities of Eclipse Transformer;
+ //this might need tailoring if the same idea gets applied to a different context.
+ ctx = new ActionContextImpl(logger,
+ new SelectionRuleImpl(logger, Collections.emptyMap(), Collections.emptyMap()),
+ new SignatureRuleImpl(logger, renames, null, null, null, null, null, Collections.emptyMap()));
+ }
+
+ byte[] transform(final String name, final byte[] bytes) {
+ logger.debug("Jakarta EE compatibility enhancer for Quarkus: transforming " + name);
+ final ClassActionImpl classTransformer = new ClassActionImpl(ctx);
+ final ByteBuffer input = ByteBuffer.wrap(bytes);
+ final ByteData inputData = new ByteDataImpl(name, input, FileUtils.DEFAULT_CHARSET);
+ final ByteData outputData = classTransformer.apply(inputData);
+ return outputData.buffer().array();
+ }
+ }
+
+}
diff --git a/extensions/aws2-kinesis/runtime/src/main/java/org/apache/camel/quarkus/component/aws2/kinesis/graalvm/Aws2KinesisSubstitutions.java b/extensions/aws2-kinesis/runtime/src/main/java/org/apache/camel/quarkus/component/aws2/kinesis/graalvm/Aws2KinesisSubstitutions.java
deleted file mode 100644
index 4bfacfe4bb4b..000000000000
--- a/extensions/aws2-kinesis/runtime/src/main/java/org/apache/camel/quarkus/component/aws2/kinesis/graalvm/Aws2KinesisSubstitutions.java
+++ /dev/null
@@ -1,165 +0,0 @@
-/*
- * 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.camel.quarkus.component.aws2.kinesis.graalvm;
-
-import java.io.UnsupportedEncodingException;
-import java.math.BigInteger;
-import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-
-import com.google.protobuf.InvalidProtocolBufferException;
-import com.oracle.svm.core.annotate.Alias;
-import com.oracle.svm.core.annotate.Substitute;
-import com.oracle.svm.core.annotate.TargetClass;
-import software.amazon.kinesis.retrieval.AggregatorUtil;
-import software.amazon.kinesis.retrieval.KinesisClientRecord;
-import software.amazon.kinesis.retrieval.kpl.Messages;
-
-public class Aws2KinesisSubstitutions {
-}
-
-/**
- * Quick (ugly) fix of https://github.com/awslabs/amazon-kinesis-client/issues/1355
- */
-@TargetClass(AggregatorUtil.class)
-final class AggregatorUtilSubstitutions {
-
- @Alias
- protected byte[] calculateTailCheck(byte[] data) {
- return null;
- }
-
- @Alias
- protected BigInteger effectiveHashKey(String partitionKey, String explicitHashKey) throws UnsupportedEncodingException {
- return null;
- }
-
- @Alias
- public KinesisClientRecord convertRecordToKinesisClientRecord(final KinesisClientRecord record,
- final boolean aggregated,
- final long subSequenceNumber,
- final String explicitHashKey) {
- return null;
- }
-
- @Substitute
- public List deaggregate(List records,
- BigInteger startingHashKey,
- BigInteger endingHashKey) {
- List result = new ArrayList<>();
- byte[] magic = new byte[AggregatorUtil.AGGREGATED_RECORD_MAGIC.length];
- byte[] digest = new byte[16];
-
- for (KinesisClientRecord r : records) {
- boolean isAggregated = true;
- long subSeqNum = 0;
- ByteBuffer bb = r.data();
-
- if (bb.remaining() >= magic.length) {
- bb.get(magic);
- } else {
- isAggregated = false;
- }
-
- if (!Arrays.equals(AggregatorUtil.AGGREGATED_RECORD_MAGIC, magic) || bb.remaining() <= 16) {
- isAggregated = false;
- }
-
- if (isAggregated) {
- int oldLimit = bb.limit();
- bb.limit(oldLimit - 16);
- byte[] messageData = new byte[bb.remaining()];
- bb.get(messageData);
- bb.limit(oldLimit);
- bb.get(digest);
- byte[] calculatedDigest = calculateTailCheck(messageData);
-
- if (!Arrays.equals(digest, calculatedDigest)) {
- isAggregated = false;
- } else {
- try {
- Messages.AggregatedRecord ar = Messages.AggregatedRecord.parseFrom(messageData);
- List pks = ar.getPartitionKeyTableList();
- List ehks = ar.getExplicitHashKeyTableList();
- long aat = r.approximateArrivalTimestamp() == null
- ? -1 : r.approximateArrivalTimestamp().toEpochMilli();
- try {
- int recordsInCurrRecord = 0;
- for (Messages.Record mr : ar.getRecordsList()) {
- String explicitHashKey = null;
- String partitionKey = pks.get((int) mr.getPartitionKeyIndex());
- if (mr.hasExplicitHashKeyIndex()) {
- explicitHashKey = ehks.get((int) mr.getExplicitHashKeyIndex());
- }
-
- BigInteger effectiveHashKey = effectiveHashKey(partitionKey, explicitHashKey);
-
- if (effectiveHashKey.compareTo(startingHashKey) < 0
- || effectiveHashKey.compareTo(endingHashKey) > 0) {
- for (int toRemove = 0; toRemove < recordsInCurrRecord; ++toRemove) {
- result.remove(result.size() - 1);
- }
- break;
- }
-
- ++recordsInCurrRecord;
-
- KinesisClientRecord record = r.toBuilder()
- .data(ByteBuffer.wrap(mr.getData().toByteArray()))
- .partitionKey(partitionKey)
- .explicitHashKey(explicitHashKey)
- .build();
- result.add(convertRecordToKinesisClientRecord(record, true, subSeqNum++, explicitHashKey));
- }
- } catch (Exception e) {
- StringBuilder sb = new StringBuilder();
- sb.append("Unexpected exception during deaggregation, record was:\n");
- sb.append("PKS:\n");
- for (String s : pks) {
- sb.append(s).append("\n");
- }
- sb.append("EHKS: \n");
- for (String s : ehks) {
- sb.append(s).append("\n");
- }
- for (Messages.Record mr : ar.getRecordsList()) {
- sb.append("Record: [hasEhk=").append(mr.hasExplicitHashKeyIndex()).append(", ")
- .append("ehkIdx=").append(mr.getExplicitHashKeyIndex()).append(", ")
- .append("pkIdx=").append(mr.getPartitionKeyIndex()).append(", ")
- .append("dataLen=").append(mr.getData().toByteArray().length).append("]\n");
- }
- sb.append("Sequence number: ").append(r.sequenceNumber()).append("\n")
- .append("Raw data: ")
- .append(jakarta.xml.bind.DatatypeConverter.printBase64Binary(messageData)).append("\n");
- // todo log.error(sb.toString(), e);
- }
- } catch (InvalidProtocolBufferException e) {
- isAggregated = false;
- }
- }
- }
-
- if (!isAggregated) {
- bb.rewind();
- result.add(r);
- }
- }
- return result;
- }
-}