-
Notifications
You must be signed in to change notification settings - Fork 2.7k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Autodetect Kafka serializer/deserializer with Reactive Messaging
- Loading branch information
Showing
10 changed files
with
2,547 additions
and
11 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
143 changes: 143 additions & 0 deletions
143
...va/io/quarkus/smallrye/reactivemessaging/kafka/deployment/DefaultSerdeDiscoveryState.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,143 @@ | ||
| package io.quarkus.smallrye.reactivemessaging.kafka.deployment; | ||
|
|
||
| import java.util.HashMap; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
| import java.util.stream.Collectors; | ||
|
|
||
| import org.eclipse.microprofile.config.ConfigProvider; | ||
| import org.jboss.jandex.AnnotationInstance; | ||
| import org.jboss.jandex.AnnotationTarget; | ||
| import org.jboss.jandex.ClassInfo; | ||
| import org.jboss.jandex.DotName; | ||
| import org.jboss.jandex.IndexView; | ||
| import org.jboss.jandex.Type; | ||
|
|
||
| import io.smallrye.reactive.messaging.kafka.KafkaConnector; | ||
|
|
||
| class DefaultSerdeDiscoveryState { | ||
| private final IndexView index; | ||
|
|
||
| private final Map<String, Boolean> isKafkaConnector = new HashMap<>(); | ||
|
|
||
| private Boolean hasConfluent; | ||
| private Boolean hasApicurio1; | ||
| private Boolean hasApicurio2; | ||
| private Boolean hasJackson; | ||
| private Boolean hasJsonb; | ||
|
|
||
| DefaultSerdeDiscoveryState(IndexView index) { | ||
| this.index = index; | ||
| } | ||
|
|
||
| boolean isKafkaConnector(boolean incoming, String channelName) { | ||
| String channelType = incoming ? "incoming" : "outgoing"; | ||
| return isKafkaConnector.computeIfAbsent(channelType + "|" + channelName, ignored -> { | ||
| String connectorKey = "mp.messaging." + channelType + "." + channelName + ".connector"; | ||
| String connector = ConfigProvider.getConfig() | ||
| .getOptionalValue(connectorKey, String.class) | ||
| .orElse("ignored"); | ||
| return KafkaConnector.CONNECTOR_NAME.equals(connector); | ||
| }); | ||
| } | ||
|
|
||
| boolean isAvroGenerated(DotName className) { | ||
| ClassInfo clazz = index.getClassByName(className); | ||
| return clazz != null && clazz.classAnnotation(DotNames.AVRO_GENERATED) != null; | ||
| } | ||
|
|
||
| boolean hasConfluent() { | ||
| if (hasConfluent == null) { | ||
| try { | ||
| Class.forName("io.confluent.kafka.serializers.KafkaAvroDeserializer", false, | ||
| Thread.currentThread().getContextClassLoader()); | ||
| hasConfluent = true; | ||
| } catch (ClassNotFoundException e) { | ||
| hasConfluent = false; | ||
| } | ||
| } | ||
|
|
||
| return hasConfluent; | ||
| } | ||
|
|
||
| boolean hasApicurio1() { | ||
| if (hasApicurio1 == null) { | ||
| try { | ||
| Class.forName("io.apicurio.registry.utils.serde.AvroKafkaDeserializer", false, | ||
| Thread.currentThread().getContextClassLoader()); | ||
| hasApicurio1 = true; | ||
| } catch (ClassNotFoundException e) { | ||
| hasApicurio1 = false; | ||
| } | ||
| } | ||
|
|
||
| return hasApicurio1; | ||
| } | ||
|
|
||
| boolean hasApicurio2() { | ||
| if (hasApicurio2 == null) { | ||
| try { | ||
| Class.forName("io.apicurio.registry.serde.avro.AvroKafkaDeserializer", false, | ||
| Thread.currentThread().getContextClassLoader()); | ||
| hasApicurio2 = true; | ||
| } catch (ClassNotFoundException e) { | ||
| hasApicurio2 = false; | ||
| } | ||
| } | ||
|
|
||
| return hasApicurio2; | ||
| } | ||
|
|
||
| boolean hasJackson() { | ||
| if (hasJackson == null) { | ||
| try { | ||
| Class.forName("com.fasterxml.jackson.databind.ObjectMapper", false, | ||
| Thread.currentThread().getContextClassLoader()); | ||
| hasJackson = true; | ||
| } catch (ClassNotFoundException e) { | ||
| hasJackson = false; | ||
| } | ||
| } | ||
|
|
||
| return hasJackson; | ||
| } | ||
|
|
||
| boolean hasJsonb() { | ||
| if (hasJsonb == null) { | ||
| try { | ||
| Class.forName("javax.json.bind.Jsonb", false, | ||
| Thread.currentThread().getContextClassLoader()); | ||
| hasJsonb = true; | ||
| } catch (ClassNotFoundException e) { | ||
| hasJsonb = false; | ||
| } | ||
| } | ||
|
|
||
| return hasJsonb; | ||
| } | ||
|
|
||
| ClassInfo getSubclassOfWithTypeArgument(DotName superclass, DotName expectedTypeArgument) { | ||
| return index.getKnownDirectSubclasses(superclass) | ||
| .stream() | ||
| .filter(it -> it.superClassType().kind() == Type.Kind.PARAMETERIZED_TYPE | ||
| && it.superClassType().asParameterizedType().arguments().size() == 1 | ||
| && it.superClassType().asParameterizedType().arguments().get(0).name().equals(expectedTypeArgument)) | ||
| .findAny() | ||
| .orElse(null); | ||
| } | ||
|
|
||
| List<AnnotationInstance> findAnnotationsOnMethods(DotName annotation) { | ||
| return index.getAnnotations(annotation) | ||
| .stream() | ||
| .filter(it -> it.target().kind() == AnnotationTarget.Kind.METHOD) | ||
| .collect(Collectors.toList()); | ||
| } | ||
|
|
||
| List<AnnotationInstance> findAnnotationsOnInjectionPoints(DotName annotation) { | ||
| return index.getAnnotations(annotation) | ||
| .stream() | ||
| .filter(it -> it.target().kind() == AnnotationTarget.Kind.FIELD | ||
| || it.target().kind() == AnnotationTarget.Kind.METHOD_PARAMETER) | ||
| .collect(Collectors.toList()); | ||
| } | ||
| } |
35 changes: 35 additions & 0 deletions
35
...oyment/src/main/java/io/quarkus/smallrye/reactivemessaging/kafka/deployment/DotNames.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,35 @@ | ||
| package io.quarkus.smallrye.reactivemessaging.kafka.deployment; | ||
|
|
||
| import org.jboss.jandex.DotName; | ||
|
|
||
| final class DotNames { | ||
| // @formatter:off | ||
| static final DotName INCOMING = DotName.createSimple(org.eclipse.microprofile.reactive.messaging.Incoming.class.getName()); | ||
| static final DotName OUTGOING = DotName.createSimple(org.eclipse.microprofile.reactive.messaging.Outgoing.class.getName()); | ||
| static final DotName CHANNEL = DotName.createSimple(org.eclipse.microprofile.reactive.messaging.Channel.class.getName()); | ||
|
|
||
| static final DotName EMITTER = DotName.createSimple(org.eclipse.microprofile.reactive.messaging.Emitter.class.getName()); | ||
| static final DotName MUTINY_EMITTER = DotName.createSimple(io.smallrye.reactive.messaging.MutinyEmitter.class.getName()); | ||
|
|
||
| static final DotName MESSAGE = DotName.createSimple(org.eclipse.microprofile.reactive.messaging.Message.class.getName()); | ||
| static final DotName KAFKA_RECORD = DotName.createSimple(io.smallrye.reactive.messaging.kafka.KafkaRecord.class.getName()); | ||
| static final DotName RECORD = DotName.createSimple(io.smallrye.reactive.messaging.kafka.Record.class.getName()); | ||
|
|
||
| static final DotName COMPLETION_STAGE = DotName.createSimple(java.util.concurrent.CompletionStage.class.getName()); | ||
| static final DotName UNI = DotName.createSimple(io.smallrye.mutiny.Uni.class.getName()); | ||
|
|
||
| static final DotName SUBSCRIBER = DotName.createSimple(org.reactivestreams.Subscriber.class.getName()); | ||
| static final DotName SUBSCRIBER_BUILDER = DotName.createSimple(org.eclipse.microprofile.reactive.streams.operators.SubscriberBuilder.class.getName()); | ||
| static final DotName PUBLISHER = DotName.createSimple(org.reactivestreams.Publisher.class.getName()); | ||
| static final DotName PUBLISHER_BUILDER = DotName.createSimple(org.eclipse.microprofile.reactive.streams.operators.PublisherBuilder.class.getName()); | ||
| static final DotName PROCESSOR = DotName.createSimple(org.reactivestreams.Processor.class.getName()); | ||
| static final DotName PROCESSOR_BUILDER = DotName.createSimple(org.eclipse.microprofile.reactive.streams.operators.ProcessorBuilder.class.getName()); | ||
| static final DotName MULTI = DotName.createSimple(io.smallrye.mutiny.Multi.class.getName()); | ||
|
|
||
| static final DotName AVRO_GENERATED = DotName.createSimple("org.apache.avro.specific.AvroGenerated"); | ||
| static final DotName OBJECT_MAPPER_DESERIALIZER = DotName.createSimple(io.quarkus.kafka.client.serialization.ObjectMapperDeserializer.class.getName()); | ||
| static final DotName OBJECT_MAPPER_SERIALIZER = DotName.createSimple(io.quarkus.kafka.client.serialization.ObjectMapperSerializer.class.getName()); | ||
| static final DotName JSONB_DESERIALIZER = DotName.createSimple(io.quarkus.kafka.client.serialization.JsonbDeserializer.class.getName()); | ||
| static final DotName JSONB_SERIALIZER = DotName.createSimple(io.quarkus.kafka.client.serialization.JsonbSerializer.class.getName()); | ||
| // @formatter:on | ||
| } |
14 changes: 14 additions & 0 deletions
14
...us/smallrye/reactivemessaging/kafka/deployment/ReactiveMessagingKafkaBuildTimeConfig.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,14 @@ | ||
| package io.quarkus.smallrye.reactivemessaging.kafka.deployment; | ||
|
|
||
| import io.quarkus.runtime.annotations.ConfigItem; | ||
| import io.quarkus.runtime.annotations.ConfigPhase; | ||
| import io.quarkus.runtime.annotations.ConfigRoot; | ||
|
|
||
| @ConfigRoot(name = "reactive-messaging.kafka", phase = ConfigPhase.BUILD_TIME) | ||
| public class ReactiveMessagingKafkaBuildTimeConfig { | ||
| /** | ||
| * Whether or not Kafka serializer/deserializer autodetection is enabled. | ||
| */ | ||
| @ConfigItem(name = "serializer-autodetection.enabled", defaultValue = "true") | ||
| public boolean serializerAutodetectionEnabled; | ||
| } |
Oops, something went wrong.