diff --git a/avro/avro-annotation-processor/build.gradle b/avro/avro-annotation-processor/build.gradle new file mode 100644 index 000000000..bda1121f3 --- /dev/null +++ b/avro/avro-annotation-processor/build.gradle @@ -0,0 +1,43 @@ +import org.apache.avro.tool.SpecificCompilerTool + +buildscript { + dependencies { + classpath libs.avro.tools + } +} + +apply from: "${project.rootDir}/gradle/in-test-generated.gradle" + +dependencies { + api project(":core:annotation-processor-common") + api project(":core:kora-app-annotation-processor") + + testImplementation project(":avro:avro-common") + testImplementation testFixtures(project(":core:annotation-processor-common")) +} + +tasks.register("generateAvroClasses") { + group("build") + + var inputDir = "$projectDir/src/test/resources/avro" + var outputDir = "$buildDir/generated/sources/avro" + inputs.dir(inputDir) + outputs.dir(outputDir) + logging.captureStandardOutput(LogLevel.INFO); + logging.captureStandardError(LogLevel.ERROR) + + doFirst { + delete outputDir + } + + doLast { + var params = ["-bigDecimal", "schema", inputDir.toString(), outputDir.toString()] + new SpecificCompilerTool().run(System.in, System.out, System.err, params) + } +} + +sourceSets.test.java { + srcDir tasks.generateAvroClasses +} + +test.dependsOn(tasks.generateAvroClasses) diff --git a/avro/avro-annotation-processor/src/main/java/io/koraframework/avro/annotation/processor/AvroAnnotationProcessor.java b/avro/avro-annotation-processor/src/main/java/io/koraframework/avro/annotation/processor/AvroAnnotationProcessor.java new file mode 100644 index 000000000..bc8b152c6 --- /dev/null +++ b/avro/avro-annotation-processor/src/main/java/io/koraframework/avro/annotation/processor/AvroAnnotationProcessor.java @@ -0,0 +1,80 @@ +package io.koraframework.avro.annotation.processor; + +import com.palantir.javapoet.ClassName; +import io.koraframework.annotation.processor.common.AbstractKoraProcessor; +import io.koraframework.annotation.processor.common.LogUtils; +import io.koraframework.annotation.processor.common.ProcessingErrorException; +import io.koraframework.avro.annotation.processor.reader.AvroReaderGenerator; +import io.koraframework.avro.annotation.processor.writer.AvroWriterGenerator; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.slf4j.event.Level; + +import javax.annotation.processing.ProcessingEnvironment; +import javax.annotation.processing.RoundEnvironment; +import javax.lang.model.element.ElementKind; +import javax.lang.model.element.TypeElement; +import java.util.List; +import java.util.Map; +import java.util.Set; + +public class AvroAnnotationProcessor extends AbstractKoraProcessor { + private static final Logger logger = LoggerFactory.getLogger(AvroAnnotationProcessor.class); + + private boolean initialized = false; + private TypeElement avroAnnotation; + private TypeElement specificRecordType; + private AvroWriterGenerator writerGenerator; + private AvroReaderGenerator readerGenerator; + + @Override + public Set getSupportedAnnotationClassNames() { + return Set.of(AvroTypes.avro); + } + + @Override + public synchronized void init(ProcessingEnvironment processingEnv) { + super.init(processingEnv); + this.avroAnnotation = processingEnv.getElementUtils().getTypeElement(AvroTypes.avro.canonicalName()); + this.specificRecordType = processingEnv.getElementUtils().getTypeElement(AvroTypes.specificRecord.canonicalName()); + if (this.avroAnnotation == null || this.specificRecordType == null) { + return; + } + + this.initialized = true; + this.writerGenerator = new AvroWriterGenerator(processingEnv); + this.readerGenerator = new AvroReaderGenerator(processingEnv); + } + + @Override + protected void process(Set annotations, RoundEnvironment roundEnv, Map> annotatedElements) { + if (!this.initialized || roundEnv.processingOver()) { + return; + } + + var avroElements = roundEnv.getElementsAnnotatedWith(this.avroAnnotation).stream() + .filter(e -> e.getKind().isClass() || e.getKind() == ElementKind.INTERFACE) + .toList(); + if (avroElements.isEmpty()) { + return; + } + + LogUtils.logElementsFull(logger, Level.DEBUG, "Generating Avro Readers & Writers for", avroElements); + for (var e : avroElements) { + if (!this.types.isAssignable(e.asType(), this.specificRecordType.asType())) { + new ProcessingErrorException("@Avro can only be used on org.apache.avro.specific.SpecificRecord types", e).printError(this.processingEnv); + continue; + } + try { + this.readerGenerator.generate((TypeElement) e); + } catch (ProcessingErrorException ex) { + ex.printError(this.processingEnv); + } + try { + this.writerGenerator.generate((TypeElement) e); + } catch (ProcessingErrorException ex) { + ex.printError(this.processingEnv); + } + } + } +} diff --git a/avro/avro-annotation-processor/src/main/java/io/koraframework/avro/annotation/processor/AvroTypes.java b/avro/avro-annotation-processor/src/main/java/io/koraframework/avro/annotation/processor/AvroTypes.java new file mode 100644 index 000000000..92abb1514 --- /dev/null +++ b/avro/avro-annotation-processor/src/main/java/io/koraframework/avro/annotation/processor/AvroTypes.java @@ -0,0 +1,21 @@ +package io.koraframework.avro.annotation.processor; + +import com.palantir.javapoet.ClassName; + +public final class AvroTypes { + + private AvroTypes() {} + + public static final ClassName avro = ClassName.get("io.koraframework.avro.common.annotation", "Avro"); + + public static final ClassName reader = ClassName.get("io.koraframework.avro.common", "AvroReader"); + public static final ClassName writer = ClassName.get("io.koraframework.avro.common", "AvroWriter"); + + public static final ClassName schema = ClassName.get("org.apache.avro", "Schema"); + public static final ClassName specificData = ClassName.get("org.apache.avro.specific", "SpecificData"); + public static final ClassName specificRecord = ClassName.get("org.apache.avro.specific", "SpecificRecord"); + public static final ClassName datumReader = ClassName.get("org.apache.avro.specific", "SpecificDatumReader"); + public static final ClassName datumWriter = ClassName.get("org.apache.avro.specific", "SpecificDatumWriter"); + public static final ClassName decoderFactory = ClassName.get("org.apache.avro.io", "DecoderFactory"); + public static final ClassName encoderFactory = ClassName.get("org.apache.avro.io", "EncoderFactory"); +} diff --git a/avro/avro-annotation-processor/src/main/java/io/koraframework/avro/annotation/processor/AvroUtils.java b/avro/avro-annotation-processor/src/main/java/io/koraframework/avro/annotation/processor/AvroUtils.java new file mode 100644 index 000000000..eaf4c9523 --- /dev/null +++ b/avro/avro-annotation-processor/src/main/java/io/koraframework/avro/annotation/processor/AvroUtils.java @@ -0,0 +1,37 @@ +package io.koraframework.avro.annotation.processor; + +import io.koraframework.annotation.processor.common.NameUtils; + +import javax.lang.model.element.Element; +import javax.lang.model.element.TypeElement; +import javax.lang.model.type.TypeMirror; +import javax.lang.model.util.Elements; +import javax.lang.model.util.Types; + + +public final class AvroUtils { + + private AvroUtils() {} + + public static String classPackage(Elements elements, Element typeElement) { + return elements.getPackageOf(typeElement).getQualifiedName().toString(); + } + + public static String writerName(Element typeElement) { + return NameUtils.generatedType(typeElement, "AvroWriter"); + } + + public static String writerName(Types types, TypeMirror typeMirror) { + var typeElement = types.asElement(typeMirror); + return writerName(typeElement); + } + + public static String readerName(TypeElement typeElement) { + return NameUtils.generatedType(typeElement, "AvroReader"); + } + + public static String readerName(Types types, TypeMirror typeMirror) { + var typeElement = types.asElement(typeMirror); + return readerName((TypeElement) typeElement); + } +} diff --git a/avro/avro-annotation-processor/src/main/java/io/koraframework/avro/annotation/processor/extension/AvroExtension.java b/avro/avro-annotation-processor/src/main/java/io/koraframework/avro/annotation/processor/extension/AvroExtension.java new file mode 100644 index 000000000..fa7978652 --- /dev/null +++ b/avro/avro-annotation-processor/src/main/java/io/koraframework/avro/annotation/processor/extension/AvroExtension.java @@ -0,0 +1,122 @@ +package io.koraframework.avro.annotation.processor.extension; + +import com.palantir.javapoet.ClassName; +import com.palantir.javapoet.CodeBlock; +import io.koraframework.annotation.processor.common.ProcessingErrorException; +import io.koraframework.avro.annotation.processor.AvroTypes; +import io.koraframework.avro.annotation.processor.AvroUtils; +import io.koraframework.avro.annotation.processor.reader.AvroReaderGenerator; +import io.koraframework.avro.annotation.processor.writer.AvroWriterGenerator; +import io.koraframework.kora.app.annotation.processor.extension.ExtensionResult; +import io.koraframework.kora.app.annotation.processor.extension.KoraExtension; +import org.jspecify.annotations.Nullable; + +import javax.annotation.processing.ProcessingEnvironment; +import javax.annotation.processing.RoundEnvironment; +import javax.lang.model.element.ElementKind; +import javax.lang.model.element.Modifier; +import javax.lang.model.element.TypeElement; +import javax.lang.model.type.DeclaredType; +import javax.lang.model.type.TypeKind; +import javax.lang.model.type.TypeMirror; +import javax.lang.model.util.Elements; +import javax.lang.model.util.Types; +import java.util.HashSet; +import java.util.List; +import java.util.Set; + +public final class AvroExtension implements KoraExtension { + + private final Types types; + private final Elements elements; + private final TypeMirror writerErasure; + private final TypeMirror readerErasure; + private final AvroReaderGenerator readerGenerator; + private final AvroWriterGenerator writerGenerator; + private final Set generatedMappers = new HashSet<>(); + + public AvroExtension(ProcessingEnvironment processingEnv) { + this.types = processingEnv.getTypeUtils(); + this.elements = processingEnv.getElementUtils(); + this.writerErasure = this.types.erasure(this.elements.getTypeElement(AvroTypes.writer.canonicalName()).asType()); + this.readerErasure = this.types.erasure(this.elements.getTypeElement(AvroTypes.reader.canonicalName()).asType()); + this.readerGenerator = new AvroReaderGenerator(processingEnv); + this.writerGenerator = new AvroWriterGenerator(processingEnv); + } + + @Nullable + @Override + public KoraExtensionDependencyGenerator getDependencyGenerator(RoundEnvironment roundEnvironment, TypeMirror typeMirror, @Nullable String tag) { + if (!isBinary(tag)) { + return null; + } + if (typeMirror.getKind() != TypeKind.DECLARED) { + return null; + } + + var erasure = this.types.erasure(typeMirror); + if (this.types.isSameType(erasure, this.writerErasure)) { + var targetElement = dependencyTarget(typeMirror); + if (targetElement == null) { + throw new ProcessingErrorException("AvroWriter can only be created for concrete org.apache.avro.specific.SpecificRecord types", this.types.asElement(typeMirror)); + } + return generateMapper(targetElement, AvroUtils.writerName(targetElement), typeMirror, tag, () -> this.writerGenerator.generate(targetElement)); + } + + if (this.types.isSameType(erasure, this.readerErasure)) { + var targetElement = dependencyTarget(typeMirror); + if (targetElement == null) { + throw new ProcessingErrorException("AvroReader can only be created for concrete org.apache.avro.specific.SpecificRecord types", this.types.asElement(typeMirror)); + } + return generateMapper(targetElement, AvroUtils.readerName(targetElement), typeMirror, tag, () -> this.readerGenerator.generate(targetElement)); + } + + return null; + } + + @Nullable + private TypeElement dependencyTarget(TypeMirror typeMirror) { + var declaredType = (DeclaredType) typeMirror; + var targetType = declaredType.getTypeArguments().get(0); + if (targetType.getKind() != TypeKind.DECLARED) { + return null; + } + var targetElement = this.types.asElement(targetType); + if (!(targetElement instanceof TypeElement typeElement) + || targetElement.getKind() == ElementKind.ENUM + || targetElement.getKind().isInterface() + || targetElement.getModifiers().contains(Modifier.ABSTRACT)) { + return null; + } + return typeElement; + } + + private KoraExtensionDependencyGenerator generateMapper( + TypeElement targetElement, + String mapperName, + TypeMirror requestedType, + @Nullable String tag, + Runnable generate + ) { + return () -> { + var packageName = AvroUtils.classPackage(this.elements, targetElement); + var mapperCanonicalName = packageName + "." + mapperName; + if (this.elements.getTypeElement(mapperCanonicalName) == null && this.generatedMappers.add(mapperCanonicalName)) { + generate.run(); + } + + return new ExtensionResult.CodeBlockResult( + targetElement, + params -> CodeBlock.of("new $T()", ClassName.get(packageName, mapperName)), + requestedType, + tag, + List.of(), + List.of() + ); + }; + } + + private boolean isBinary(@Nullable String tag) { + return tag == null || tag.equals(AvroTypes.avro.canonicalName()); + } +} diff --git a/avro/avro-annotation-processor/src/main/java/io/koraframework/avro/annotation/processor/extension/AvroExtensionFactory.java b/avro/avro-annotation-processor/src/main/java/io/koraframework/avro/annotation/processor/extension/AvroExtensionFactory.java new file mode 100644 index 000000000..2f090a3cb --- /dev/null +++ b/avro/avro-annotation-processor/src/main/java/io/koraframework/avro/annotation/processor/extension/AvroExtensionFactory.java @@ -0,0 +1,21 @@ +package io.koraframework.avro.annotation.processor.extension; + +import io.koraframework.avro.annotation.processor.AvroTypes; +import io.koraframework.kora.app.annotation.processor.extension.ExtensionFactory; +import io.koraframework.kora.app.annotation.processor.extension.KoraExtension; + +import javax.annotation.processing.ProcessingEnvironment; +import java.util.Optional; + +public class AvroExtensionFactory implements ExtensionFactory { + + @Override + public Optional create(ProcessingEnvironment processingEnvironment) { + var avro = processingEnvironment.getElementUtils().getTypeElement(AvroTypes.avro.canonicalName()); + if (avro == null) { + return Optional.empty(); + } else { + return Optional.of(new AvroExtension(processingEnvironment)); + } + } +} diff --git a/avro/avro-annotation-processor/src/main/java/io/koraframework/avro/annotation/processor/reader/AvroReaderGenerator.java b/avro/avro-annotation-processor/src/main/java/io/koraframework/avro/annotation/processor/reader/AvroReaderGenerator.java new file mode 100644 index 000000000..d0dc7d60a --- /dev/null +++ b/avro/avro-annotation-processor/src/main/java/io/koraframework/avro/annotation/processor/reader/AvroReaderGenerator.java @@ -0,0 +1,69 @@ +package io.koraframework.avro.annotation.processor.reader; + +import com.palantir.javapoet.*; +import io.koraframework.annotation.processor.common.CommonClassNames; +import io.koraframework.annotation.processor.common.CommonUtils; +import io.koraframework.avro.annotation.processor.AvroTypes; +import io.koraframework.avro.annotation.processor.AvroUtils; +import org.jspecify.annotations.Nullable; + +import javax.annotation.processing.ProcessingEnvironment; +import javax.lang.model.element.Modifier; +import javax.lang.model.element.TypeElement; +import javax.lang.model.element.TypeParameterElement; +import java.io.IOException; +import java.io.InputStream; + +public class AvroReaderGenerator { + private final ProcessingEnvironment env; + + public AvroReaderGenerator(ProcessingEnvironment processingEnvironment) { + this.env = processingEnvironment; + } + + public void generate(TypeElement element) { + var typeName = TypeName.get(element.asType()); + var typeBuilder = TypeSpec.classBuilder(AvroUtils.readerName(element)) + .addAnnotation(AnnotationSpec.builder(CommonClassNames.koraGenerated) + .addMember("value", CodeBlock.of("$S", AvroReaderGenerator.class.getCanonicalName())) + .build()) + .addAnnotation(AvroTypes.avro) + .addSuperinterface(ParameterizedTypeName.get(AvroTypes.reader, typeName)) + .addModifiers(Modifier.PUBLIC, Modifier.FINAL) + .addOriginatingElement(element); + + for (TypeParameterElement typeParameter : element.getTypeParameters()) { + typeBuilder.addTypeVariable(TypeVariableName.get(typeParameter)); + } + + typeBuilder.addField(FieldSpec.builder(AvroTypes.schema, "SCHEMA") + .addModifiers(Modifier.PRIVATE, Modifier.STATIC, Modifier.FINAL) + .initializer("$T.getClassSchema()", typeName) + .build()); + typeBuilder.addField(FieldSpec.builder(AvroTypes.specificData, "SPECIFIC_DATA") + .addModifiers(Modifier.PRIVATE, Modifier.STATIC, Modifier.FINAL) + .initializer("new $T().getSpecificData()", typeName) + .build()); + typeBuilder.addField(FieldSpec.builder(ParameterizedTypeName.get(AvroTypes.datumReader, typeName), "READER") + .addModifiers(Modifier.PRIVATE, Modifier.STATIC, Modifier.FINAL) + .initializer("new $T<>(SCHEMA, SCHEMA, SPECIFIC_DATA)", AvroTypes.datumReader) + .build()); + + var method = MethodSpec.methodBuilder("read") + .addModifiers(Modifier.PUBLIC, Modifier.FINAL) + .addException(IOException.class) + .addAnnotation(Nullable.class) + .addAnnotation(Override.class) + .addParameter(TypeName.get(InputStream.class), "value") + .returns(typeName); + method.beginControlFlow("if (value == null || value.available() == 0)"); + method.addStatement("return null"); + method.endControlFlow(); + method.addStatement("var decoder = $T.get().directBinaryDecoder(value, null)", AvroTypes.decoderFactory); + method.addStatement("return READER.read(new $T(), decoder)", typeName); + + typeBuilder.addMethod(method.build()); + var javaFile = JavaFile.builder(AvroUtils.classPackage(this.env.getElementUtils(), element), typeBuilder.build()).build(); + CommonUtils.safeWriteTo(this.env, javaFile); + } +} diff --git a/avro/avro-annotation-processor/src/main/java/io/koraframework/avro/annotation/processor/writer/AvroWriterGenerator.java b/avro/avro-annotation-processor/src/main/java/io/koraframework/avro/annotation/processor/writer/AvroWriterGenerator.java new file mode 100644 index 000000000..2e87364ab --- /dev/null +++ b/avro/avro-annotation-processor/src/main/java/io/koraframework/avro/annotation/processor/writer/AvroWriterGenerator.java @@ -0,0 +1,75 @@ +package io.koraframework.avro.annotation.processor.writer; + +import com.palantir.javapoet.*; +import io.koraframework.annotation.processor.common.CommonClassNames; +import io.koraframework.annotation.processor.common.CommonUtils; +import io.koraframework.avro.annotation.processor.AvroTypes; +import io.koraframework.avro.annotation.processor.AvroUtils; +import org.jspecify.annotations.Nullable; + +import javax.annotation.processing.ProcessingEnvironment; +import javax.lang.model.element.Modifier; +import javax.lang.model.element.TypeElement; +import java.io.ByteArrayOutputStream; +import java.io.IOException; + +public class AvroWriterGenerator { + private final ProcessingEnvironment env; + + public AvroWriterGenerator(ProcessingEnvironment processingEnvironment) { + this.env = processingEnvironment; + } + + public void generate(TypeElement element) { + var typeName = TypeName.get(element.asType()); + var typeBuilder = TypeSpec.classBuilder(AvroUtils.writerName(element)) + .addAnnotation(AnnotationSpec.builder(CommonClassNames.koraGenerated) + .addMember("value", CodeBlock.of("$S", AvroWriterGenerator.class.getCanonicalName())) + .build()) + .addAnnotation(AvroTypes.avro) + .addSuperinterface(ParameterizedTypeName.get(AvroTypes.writer, typeName)) + .addModifiers(Modifier.PUBLIC, Modifier.FINAL) + .addOriginatingElement(element); + + for (var typeParameter : element.getTypeParameters()) { + typeBuilder.addTypeVariable(TypeVariableName.get(typeParameter)); + } + + typeBuilder.addField(FieldSpec.builder(ArrayTypeName.of(TypeName.BYTE), "EMPTY") + .addModifiers(Modifier.PRIVATE, Modifier.STATIC, Modifier.FINAL) + .initializer("new byte[]{}") + .build()); + typeBuilder.addField(FieldSpec.builder(AvroTypes.schema, "SCHEMA") + .addModifiers(Modifier.PRIVATE, Modifier.STATIC, Modifier.FINAL) + .initializer("$T.getClassSchema()", typeName) + .build()); + typeBuilder.addField(FieldSpec.builder(AvroTypes.specificData, "SPECIFIC_DATA") + .addModifiers(Modifier.PRIVATE, Modifier.STATIC, Modifier.FINAL) + .initializer("new $T().getSpecificData()", typeName) + .build()); + typeBuilder.addField(FieldSpec.builder(ParameterizedTypeName.get(AvroTypes.datumWriter, typeName), "WRITER") + .addModifiers(Modifier.PRIVATE, Modifier.STATIC, Modifier.FINAL) + .initializer("new $T<>(SCHEMA, SPECIFIC_DATA)", AvroTypes.datumWriter) + .build()); + + var method = MethodSpec.methodBuilder("writeBytes") + .addModifiers(Modifier.PUBLIC, Modifier.FINAL) + .addException(IOException.class) + .addAnnotation(Override.class) + .addParameter(ParameterSpec.builder(typeName, "value").addAnnotation(Nullable.class).build()) + .returns(ArrayTypeName.of(TypeName.BYTE)); + method.beginControlFlow("if (value == null)"); + method.addStatement("return EMPTY"); + method.endControlFlow(); + method.beginControlFlow("try (var os = new $T())", ByteArrayOutputStream.class); + method.addStatement("var encoder = $T.get().directBinaryEncoder(os, null)", AvroTypes.encoderFactory); + method.addStatement("WRITER.write(value, encoder)"); + method.addStatement("encoder.flush()"); + method.addStatement("return os.toByteArray()"); + method.endControlFlow(); + + typeBuilder.addMethod(method.build()); + var javaFile = JavaFile.builder(AvroUtils.classPackage(this.env.getElementUtils(), element), typeBuilder.build()).build(); + CommonUtils.safeWriteTo(this.env, javaFile); + } +} diff --git a/avro/avro-annotation-processor/src/main/resources/META-INF/services/io.koraframework.kora.app.annotation.processor.extension.ExtensionFactory b/avro/avro-annotation-processor/src/main/resources/META-INF/services/io.koraframework.kora.app.annotation.processor.extension.ExtensionFactory new file mode 100644 index 000000000..91c83b6f2 --- /dev/null +++ b/avro/avro-annotation-processor/src/main/resources/META-INF/services/io.koraframework.kora.app.annotation.processor.extension.ExtensionFactory @@ -0,0 +1 @@ +io.koraframework.avro.annotation.processor.extension.AvroExtensionFactory diff --git a/avro/avro-annotation-processor/src/main/resources/META-INF/services/javax.annotation.processing.Processor b/avro/avro-annotation-processor/src/main/resources/META-INF/services/javax.annotation.processing.Processor new file mode 100644 index 000000000..7593eb027 --- /dev/null +++ b/avro/avro-annotation-processor/src/main/resources/META-INF/services/javax.annotation.processing.Processor @@ -0,0 +1 @@ +io.koraframework.avro.annotation.processor.AvroAnnotationProcessor diff --git a/avro/avro-annotation-processor/src/test/java/io/koraframework/avro/annotation/processor/AbstractAvroAnnotationProcessorTest.java b/avro/avro-annotation-processor/src/test/java/io/koraframework/avro/annotation/processor/AbstractAvroAnnotationProcessorTest.java new file mode 100644 index 000000000..94c692c54 --- /dev/null +++ b/avro/avro-annotation-processor/src/test/java/io/koraframework/avro/annotation/processor/AbstractAvroAnnotationProcessorTest.java @@ -0,0 +1,243 @@ +package io.koraframework.avro.annotation.processor; + +import io.koraframework.annotation.processor.common.AbstractAnnotationProcessorTest; +import io.koraframework.avro.common.AvroReader; +import io.koraframework.avro.common.AvroWriter; +import org.apache.avro.generic.IndexedRecord; +import org.apache.avro.io.DecoderFactory; +import org.apache.avro.io.Encoder; +import org.apache.avro.io.EncoderFactory; +import org.apache.avro.specific.SpecificData; +import org.apache.avro.specific.SpecificDatumReader; +import org.apache.avro.specific.SpecificDatumWriter; +import org.intellij.lang.annotations.Language; +import org.jetbrains.annotations.Nullable; + +import java.io.ByteArrayOutputStream; +import java.io.File; +import java.io.IOException; +import java.io.InputStream; +import java.lang.reflect.Field; +import java.lang.reflect.InvocationTargetException; +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.time.Instant; +import java.util.Base64; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; + +public abstract class AbstractAvroAnnotationProcessorTest extends AbstractAnnotationProcessorTest { + + @Override + protected String commonImports() { + return super.commonImports() + """ + import io.koraframework.common.annotation.KoraApp; + import io.koraframework.avro.common.annotation.*; + import io.koraframework.avro.common.AvroReader; + import io.koraframework.avro.common.AvroWriter; + import iokoraframework.kora.avro.TestAvro; + import java.util.Optional; + """; + } + + protected IndexedRecord getTestAvroGenerated() { + return getTestAvroGeneratedRecord(); + } + + protected IndexedRecord getTestAvroGeneratedRecord() { + return iokoraframework.kora.avro.TestAvro.newBuilder() + .setCluster("cluster") + .setDate(Instant.EPOCH) + .setDescription("descr") + .setCounter(12345L) + .setFlag(true) + .build(); + } + + protected byte[] getTestAvroAsBytes() { + return Base64.getDecoder().decode("DmNsdXN0ZXICAAIKZGVzY3IC8sABAgE="); + } + + protected void assertThatTestAvroValid(IndexedRecord expected, IndexedRecord actual) { + assertThat(actual).isNotNull(); + assertThat(actual.get(0).toString()).isEqualTo(expected.get(0).toString()); + assertThat(actual.get(1)).isEqualTo(expected.get(1)); + assertThat(actual.get(2).toString()).isEqualTo(expected.get(2).toString()); + assertThat(actual.get(3)).isEqualTo(expected.get(3)); + assertThat(actual.get(4)).isEqualTo(expected.get(4)); + } + + protected String getAvroClass() { + return getAvroClass(null); + } + + protected String getAvroClass(@Nullable String annotation) { + try { + var path = "build/generated/sources/avro/iokoraframework/kora/avro/TestAvro.java"; + List strings = Files.lines(new File(path).toPath()) + .map(s -> s.replace("iokoraframework.kora.", "")) + .toList(); + String avro = String.join("\n", strings.subList(7, strings.size())); + avro = avro.replace("iokoraframework.kora", testPackage()); + if (annotation != null) { + avro = avro.replaceFirst("@org\\.apache\\.avro\\.specific\\.AvroGenerated", annotation + "\n@org.apache.avro.specific.AvroGenerated"); + } + return avro; + } catch (IOException e) { + throw new RuntimeException(e); + } + } + + protected void compile(@Language("java") String... sources) { + var compileResult = compile(List.of(new AvroAnnotationProcessor()), sources); + if (compileResult.isFailed()) { + throw compileResult.compilationException(); + } + } + + @SuppressWarnings("unchecked") + protected AvroReader reader(Class forClass, Object... params) { + try { + return (AvroReader) this.compileResult.loadClass("$" + forClass + "_AvroReader") + .getConstructors()[0] + .newInstance(params); + } catch (InstantiationException | IllegalAccessException | InvocationTargetException e) { + throw new RuntimeException(e); + } + } + + protected AvroReader reader(String forClass, Object... params) { + return reader(testPackage(), forClass, params); + } + + @SuppressWarnings("unchecked") + protected AvroReader reader(String packageName, String forClass, Object... params) { + try { + return (AvroReader) this.compileResult.cl().loadClass(packageName + ".$" + forClass + "_AvroReader") + .getConstructors()[0] + .newInstance(params); + } catch (InstantiationException | IllegalAccessException | InvocationTargetException | ClassNotFoundException e) { + throw new RuntimeException(e); + } + } + + protected AvroWriter writer(String forClass, Object... params) { + return writer(testPackage(), forClass, params); + } + + @SuppressWarnings("unchecked") + protected AvroWriter writer(String packageName, String forClass, Object... params) { + try { + return (AvroWriter) this.compileResult.cl().loadClass(packageName + ".$" + forClass + "_AvroWriter") + .getConstructors()[0] + .newInstance(params); + } catch (InstantiationException | IllegalAccessException | InvocationTargetException | ClassNotFoundException e) { + throw new RuntimeException(e); + } + } + + protected ReaderAndWriter mapper(String forClass) { + return mapper(forClass, List.of(), List.of()); + } + + protected ReaderAndWriter mapper(String forClass, List readerParams, List writerParams) { + AvroReader reader = reader(forClass, readerParams.toArray()); + AvroWriter writer = writer(forClass, writerParams.toArray()); + return new ReaderAndWriter<>(reader, writer); + } + + protected static class ReaderAndWriter implements AvroReader, AvroWriter { + private final AvroReader reader; + private final AvroWriter writer; + + protected ReaderAndWriter(AvroReader reader, AvroWriter writer) { + this.reader = reader; + this.writer = writer; + } + + @Nullable + @Override + public T read(ByteBuffer buffer) throws IOException { + return reader.read(buffer); + } + + @Nullable + @Override + public T read(byte[] bytes) throws IOException { + return reader.read(bytes); + } + + @Nullable + @Override + public T read(byte[] bytes, int offset, int length) throws IOException { + return reader.read(bytes, offset, length); + } + + @Nullable + @Override + public T read(InputStream is) throws IOException { + return reader.read(is); + } + + @Override + public byte[] writeBytes(@Nullable T value) throws IOException { + return writer.writeBytes(value); + } + + public void verify(T expectedObject, String expectedAvro) { + verifyRead(expectedAvro, expectedObject); + verifyWrite(expectedObject, expectedAvro); + } + + public void verifyRead(String expectedAvro, T expectedObject) { + try { + var object = this.reader.read(expectedAvro.getBytes(StandardCharsets.UTF_8)); + assertThat(object).isEqualTo(expectedObject); + } catch (IOException e) { + throw new RuntimeException(e); + } + } + + public void verifyWrite(T expectedObject, String expectedAvro) { + try { + var avro = this.writer.writeBytes(expectedObject); + assertThat(avro).asString(StandardCharsets.UTF_8).isEqualTo(expectedAvro); + } catch (IOException e) { + throw new RuntimeException(e); + } + } + } + + protected byte[] writeAsBinary(IndexedRecord value) { + try (var stream = new ByteArrayOutputStream()) { + Field fieldData = value.getClass().getDeclaredField("MODEL$"); + fieldData.setAccessible(true); + SpecificData data = (SpecificData) fieldData.get(value); + + var writer = new SpecificDatumWriter<>(value.getSchema(), data); + Encoder encoder = EncoderFactory.get().directBinaryEncoder(stream, null); + writer.write(value, encoder); + encoder.flush(); + return stream.toByteArray(); + } catch (Exception e) { + throw new IllegalStateException(e); + } + } + + protected IndexedRecord readAsBinary(byte[] value) { + try { + var generated = getTestAvroGeneratedRecord(); + Field fieldData = generated.getClass().getDeclaredField("MODEL$"); + fieldData.setAccessible(true); + SpecificData data = (SpecificData) fieldData.get(generated); + + var reader = new SpecificDatumReader<>(generated.getSchema(), generated.getSchema(), data); + var binaryDecoder = DecoderFactory.get().binaryDecoder(value, null); + return (IndexedRecord) reader.read(null, binaryDecoder); + } catch (Exception e) { + throw new IllegalStateException(e); + } + } +} diff --git a/avro/avro-annotation-processor/src/test/java/io/koraframework/avro/annotation/processor/AvroTests.java b/avro/avro-annotation-processor/src/test/java/io/koraframework/avro/annotation/processor/AvroTests.java new file mode 100644 index 000000000..0bddeb0aa --- /dev/null +++ b/avro/avro-annotation-processor/src/test/java/io/koraframework/avro/annotation/processor/AvroTests.java @@ -0,0 +1,97 @@ +package io.koraframework.avro.annotation.processor; + +import org.apache.avro.generic.IndexedRecord; +import org.junit.jupiter.api.Test; +import io.koraframework.kora.app.annotation.processor.KoraAppProcessor; + +import java.io.IOException; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; + +public class AvroTests extends AbstractAvroAnnotationProcessorTest { + + @Test + public void testReaderFromExtension() { + compile(List.of(new AvroAnnotationProcessor(), new KoraAppProcessor()), + """ + @KoraApp + public interface TestApp { + @Root + default String root(AvroReader r) {return "";} + } + """); + + compileResult.assertSuccess(); + var reader = reader("iokoraframework.kora.avro", "TestAvro"); + assertThat(reader).isNotNull(); + + var testAvro = getTestAvroGeneratedRecord(); + byte[] bytes = getTestAvroAsBytes(); + IndexedRecord read = reader.readUnchecked(bytes); + assertThatTestAvroValid(testAvro, read); + } + + @Test + public void testReaderTaggedFromExtension() { + compile(List.of(new AvroAnnotationProcessor(), new KoraAppProcessor()), + """ + @KoraApp + public interface TestApp { + @Root + default String root(@Avro AvroReader r) {return "";} + } + """); + + compileResult.assertSuccess(); + var reader = reader("iokoraframework.kora.avro", "TestAvro"); + assertThat(reader).isNotNull(); + + var testAvro = getTestAvroGeneratedRecord(); + byte[] bytes = getTestAvroAsBytes(); + IndexedRecord read = reader.readUnchecked(bytes); + assertThatTestAvroValid(testAvro, read); + } + + @Test + public void testWriterFromExtension() throws IOException { + compile(List.of(new AvroAnnotationProcessor(), new KoraAppProcessor()), + """ + @KoraApp + public interface TestApp { + @Root + default String root(AvroWriter r) {return "";} + } + """); + + compileResult.assertSuccess(); + var writer = writer("iokoraframework.kora.avro", "TestAvro"); + assertThat(writer).isNotNull(); + + IndexedRecord testAvro = getTestAvroGeneratedRecord(); + byte[] bytes = writer.writeBytesUnchecked(testAvro); + IndexedRecord restored = readAsBinary(bytes); + assertThatTestAvroValid(testAvro, restored); + } + + @Test + public void testWriterTaggedFromExtension() { + compile(List.of(new AvroAnnotationProcessor(), new KoraAppProcessor()), + """ + @KoraApp + public interface TestApp { + @Root + default String root(@Avro AvroWriter r) {return "";} + } + """); + + compileResult.assertSuccess(); + var writer = writer("iokoraframework.kora.avro", "TestAvro"); + assertThat(writer).isNotNull(); + + IndexedRecord testAvro = getTestAvroGeneratedRecord(); + byte[] bytes = writer.writeBytesUnchecked(testAvro); + IndexedRecord restored = readAsBinary(bytes); + assertThatTestAvroValid(testAvro, restored); + } +} diff --git a/avro/avro-annotation-processor/src/test/resources/avro/TestAvro.avsc b/avro/avro-annotation-processor/src/test/resources/avro/TestAvro.avsc new file mode 100644 index 000000000..4c90c022b --- /dev/null +++ b/avro/avro-annotation-processor/src/test/resources/avro/TestAvro.avsc @@ -0,0 +1,43 @@ +{ + "type": "record", + "name": "TestAvro", + "namespace": "iokoraframework.kora.avro", + "fields": [ + { + "name": "cluster", + "type": "string" + }, + { + "name": "date", + "type": [ + "null", + { + "type": "long", + "logicalType": "timestamp-millis" + } + ] + }, + { + "name": "description", + "type": [ + "null", + "string" + ] + }, + { + "name": "counter", + "type": [ + "null", + "long" + ] + }, + { + "name": "flag", + "type": [ + "null", + "boolean" + ], + "default": null + } + ] +} diff --git a/avro/avro-common/build.gradle b/avro/avro-common/build.gradle new file mode 100644 index 000000000..d028c26df --- /dev/null +++ b/avro/avro-common/build.gradle @@ -0,0 +1,5 @@ +dependencies { + api project(":core:common") + + api libs.avro +} diff --git a/avro/avro-common/src/main/java/io/koraframework/avro/common/AvroReader.java b/avro/avro-common/src/main/java/io/koraframework/avro/common/AvroReader.java new file mode 100644 index 000000000..d3f83938c --- /dev/null +++ b/avro/avro-common/src/main/java/io/koraframework/avro/common/AvroReader.java @@ -0,0 +1,75 @@ +package io.koraframework.avro.common; + +import org.apache.avro.generic.IndexedRecord; +import org.jspecify.annotations.Nullable; + +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.UncheckedIOException; +import java.nio.ByteBuffer; + +/** + * Русский: Контракт читателя AVRO со всеми методами чтения + *
+ * English: AVRO reader contract with all read methods + */ +public interface AvroReader { + + @Nullable + T read(InputStream is) throws IOException; + + @Nullable + default T read(byte[] bytes) throws IOException { + try (var is = new ByteArrayInputStream(bytes)) { + return read(is); + } + } + + @Nullable + default T read(byte[] bytes, int offset, int length) throws IOException { + try (var is = new ByteArrayInputStream(bytes, offset, length)) { + return read(is); + } + } + + @Nullable + default T read(ByteBuffer buffer) throws IOException { + if (buffer.hasArray()) { + try (var is = new ByteArrayInputStream(buffer.array(), buffer.arrayOffset() + buffer.position(), buffer.remaining())) { + return read(is); + } + } else { + try (var is = new ByteBufferInputStream(buffer)) { + return read(is); + } + } + } + + @Nullable + default T readUnchecked(byte[] bytes) throws UncheckedIOException { + try { + return read(bytes); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } + + @Nullable + default T readUnchecked(byte[] bytes, int offset, int length) throws UncheckedIOException { + try { + return read(bytes, offset, length); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } + + @Nullable + default T readUnchecked(InputStream is) throws UncheckedIOException { + try { + return read(is); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } +} diff --git a/avro/avro-common/src/main/java/io/koraframework/avro/common/AvroWriter.java b/avro/avro-common/src/main/java/io/koraframework/avro/common/AvroWriter.java new file mode 100644 index 000000000..eb85ed4ae --- /dev/null +++ b/avro/avro-common/src/main/java/io/koraframework/avro/common/AvroWriter.java @@ -0,0 +1,25 @@ +package io.koraframework.avro.common; + +import org.apache.avro.generic.IndexedRecord; +import org.jspecify.annotations.Nullable; + +import java.io.IOException; +import java.io.UncheckedIOException; + +/** + * Русский: Контракт писателя AVRO со всеми методами записи + *
+ * English: AVRO writer contract with all write methods + */ +public interface AvroWriter { + + byte[] writeBytes(T value) throws IOException; + + default byte[] writeBytesUnchecked(T value) throws UncheckedIOException { + try { + return writeBytes(value); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } +} diff --git a/avro/avro-common/src/main/java/io/koraframework/avro/common/ByteBufferInputStream.java b/avro/avro-common/src/main/java/io/koraframework/avro/common/ByteBufferInputStream.java new file mode 100644 index 000000000..a2bb028e6 --- /dev/null +++ b/avro/avro-common/src/main/java/io/koraframework/avro/common/ByteBufferInputStream.java @@ -0,0 +1,124 @@ +package io.koraframework.avro.common; + +import java.io.InputStream; +import java.nio.ByteBuffer; +import java.util.Objects; + +final class ByteBufferInputStream extends InputStream { + + /** + * The input ByteBuffer that was provided. + * The ByteBuffer should be supplied with position and limit correctly set as appropriate + */ + private final ByteBuffer backendBuffer; + + public ByteBufferInputStream(ByteBuffer backendBuffer) { + Objects.requireNonNull(backendBuffer, "Given backend buffer can not be null!"); + this.backendBuffer = backendBuffer; + this.backendBuffer.mark(); // to prevent java.nio.InvalidMarkException on InputStream.reset() if mark had not been set + } + + /** + * Reads the next byte of data from this ByteBuffer. The value byte is returned as an int in the range 0-255. + * If no byte is available because the end of the buffer has been reached, the value -1 is returned. + * + * @return the next byte of data, or -1 if the limit/end of the buffer has been reached. + */ + public int read() { + return backendBuffer.hasRemaining() + ? (backendBuffer.get() & 0xff) + : -1; + } + + /** + * Reads up to len bytes of data into an array of bytes from this ByteBuffer. + * If the buffer has no remaining bytes, then -1 is returned to indicate end of file. + * Otherwise, the number k of bytes read is equal to the smaller of len and buffer remaining. + * + * @param b the buffer into which the data is read. + * @param off the start offset in the destination array b + * @param len the maximum number of bytes read. + * @return the total number of bytes read into the buffer, or -1 if there is no more data because the limit/end of + * the ByteBuffer has been reached. + * @throws NullPointerException If b is null. + * @throws IndexOutOfBoundsException If off is negative, len is negative, or len is greater than b.length - off + */ + public int read(byte b[], int off, int len) { + if (b == null) { + throw new NullPointerException(); + } else if (off < 0 || len < 0 || len > b.length - off) { + throw new IndexOutOfBoundsException(); + } + + if (!backendBuffer.hasRemaining()) { + return -1; + } + + int remaining = backendBuffer.remaining(); + if (len > remaining) { + len = remaining; + } + + if (len <= 0) { + return 0; + } + backendBuffer.get(b, off, len); + return len; + } + + /** + * Skips n bytes of input from this ByteBuffer. Fewer bytes might be skipped if the limit is reached. + * + * @param n the number of bytes to be skipped. + * @return the actual number of bytes skipped. + */ + public long skip(long n) { + int skipAmount = (n < 0) + ? 0 + : ((n > Integer.MAX_VALUE) ? Integer.MAX_VALUE : (int) n); + + if (skipAmount > backendBuffer.remaining()) { + skipAmount = backendBuffer.remaining(); + } + + int newPos = backendBuffer.position() + skipAmount; + backendBuffer.position(newPos); + return skipAmount; + } + + /** + * Returns remaining bytes available in this ByteBuffer + * + * @return the number of remaining bytes that can be read (or skipped over) from this ByteBuffer. + */ + public int available() { + return backendBuffer.remaining(); + } + + public boolean markSupported() { + return true; + } + + /** + * Set the current marked position in the ByteBuffer. + *

Note: The readAheadLimit for this class has no meaning. + */ + public void mark(int readAheadLimit) { + backendBuffer.mark(); + } + + /** + * Resets the ByteBuffer to the marked position. + */ + public void reset() { + backendBuffer.reset(); + } + + /** + * Closing a ByteBuffer has no effect. + * The methods in this class can be called after the stream has been closed without generating an IOException. + */ + public void close() { + + } +} diff --git a/avro/avro-common/src/main/java/io/koraframework/avro/common/GenericAvroReader.java b/avro/avro-common/src/main/java/io/koraframework/avro/common/GenericAvroReader.java new file mode 100644 index 000000000..8a6270f89 --- /dev/null +++ b/avro/avro-common/src/main/java/io/koraframework/avro/common/GenericAvroReader.java @@ -0,0 +1,29 @@ +package io.koraframework.avro.common; + +import org.apache.avro.Schema; +import org.apache.avro.generic.GenericDatumReader; +import org.apache.avro.generic.GenericRecord; +import org.apache.avro.io.DecoderFactory; +import org.jspecify.annotations.Nullable; + +import java.io.IOException; +import java.io.InputStream; + +public final class GenericAvroReader implements AvroReader { + + private final GenericDatumReader reader; + + public GenericAvroReader(Schema schema) { + this.reader = new GenericDatumReader<>(schema); + } + + @Nullable + @Override + public GenericRecord read(InputStream is) throws IOException { + if (is == null) { + return null; + } + var decoder = DecoderFactory.get().directBinaryDecoder(is, null); + return this.reader.read(null, decoder); + } +} diff --git a/avro/avro-common/src/main/java/io/koraframework/avro/common/GenericAvroWriter.java b/avro/avro-common/src/main/java/io/koraframework/avro/common/GenericAvroWriter.java new file mode 100644 index 000000000..2c18b8e84 --- /dev/null +++ b/avro/avro-common/src/main/java/io/koraframework/avro/common/GenericAvroWriter.java @@ -0,0 +1,34 @@ +package io.koraframework.avro.common; + +import org.apache.avro.Schema; +import org.apache.avro.generic.GenericDatumWriter; +import org.apache.avro.generic.GenericRecord; +import org.apache.avro.io.EncoderFactory; +import org.jspecify.annotations.Nullable; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; + +public final class GenericAvroWriter implements AvroWriter { + + private static final byte[] EMPTY = new byte[]{}; + + private final GenericDatumWriter writer; + + public GenericAvroWriter(Schema schema) { + this.writer = new GenericDatumWriter<>(schema); + } + + @Override + public byte[] writeBytes(@Nullable GenericRecord value) throws IOException { + if (value == null) { + return EMPTY; + } + try (var os = new ByteArrayOutputStream()) { + var encoder = EncoderFactory.get().directBinaryEncoder(os, null); + this.writer.write(value, encoder); + encoder.flush(); + return os.toByteArray(); + } + } +} diff --git a/avro/avro-common/src/main/java/io/koraframework/avro/common/annotation/Avro.java b/avro/avro-common/src/main/java/io/koraframework/avro/common/annotation/Avro.java new file mode 100644 index 000000000..493371565 --- /dev/null +++ b/avro/avro-common/src/main/java/io/koraframework/avro/common/annotation/Avro.java @@ -0,0 +1,13 @@ +package io.koraframework.avro.common.annotation; + +import io.koraframework.common.annotation.Tag; + +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +@Tag(Avro.class) +@Target({ElementType.TYPE, ElementType.PARAMETER, ElementType.METHOD, ElementType.TYPE_USE}) +@Retention(RetentionPolicy.RUNTIME) +public @interface Avro {} diff --git a/avro/avro-common/src/main/java/module-info.java b/avro/avro-common/src/main/java/module-info.java new file mode 100644 index 000000000..97541a971 --- /dev/null +++ b/avro/avro-common/src/main/java/module-info.java @@ -0,0 +1,11 @@ +import org.jspecify.annotations.NullMarked; + +@NullMarked +module kora.avro.common { + exports io.koraframework.avro.common; + exports io.koraframework.avro.common.annotation; + + requires transitive kora.common; + + requires transitive org.apache.avro; +} diff --git a/avro/avro-symbol-processor/build.gradle b/avro/avro-symbol-processor/build.gradle new file mode 100644 index 000000000..4d1fc0a3f --- /dev/null +++ b/avro/avro-symbol-processor/build.gradle @@ -0,0 +1,51 @@ +import org.apache.avro.tool.SpecificCompilerTool + +buildscript { + dependencies { + classpath libs.avro.tools + } +} + +apply from: "${project.rootDir}/gradle/kotlin-plugin.gradle" +apply from: "${project.rootDir}/gradle/in-test-generated.gradle" + +dependencies { + api project(':core:symbol-processor-common') + api project(':core:kora-app-symbol-processor') + + testImplementation project(':avro:avro-common') + testImplementation testFixtures(project(':core:symbol-processor-common')) + + testImplementation libs.classgraph + testImplementation libs.kotlin.compiler + testImplementation libs.kotlin.compiler + testImplementation libs.junit.jupiter + testImplementation libs.ksp.api +} + +tasks.register("generateAvroClasses") { + group("build") + + var inputDir = "$projectDir/src/test/resources/avro" + var outputDir = "$buildDir/generated/sources/avro" + inputs.dir(inputDir) + outputs.dir(outputDir) + logging.captureStandardOutput(LogLevel.INFO) + logging.captureStandardError(LogLevel.ERROR) + + doFirst { + delete outputDir + delete "$buildDir/in-tesVt-generated-ksp/sources/iokoraframework/kora" + } + + doLast { + var params = ["-bigDecimal", "schema", inputDir.toString(), outputDir.toString()] + new SpecificCompilerTool().run(System.in, System.out, System.err, params) + } +} + +sourceSets.test.java { + srcDir tasks.generateAvroClasses +} + +test.dependsOn(tasks.generateAvroClasses) diff --git a/avro/avro-symbol-processor/src/main/kotlin/io/koraframework/avro/symbol/processor/AvroSymbolProcessor.kt b/avro/avro-symbol-processor/src/main/kotlin/io/koraframework/avro/symbol/processor/AvroSymbolProcessor.kt new file mode 100644 index 000000000..7277c434a --- /dev/null +++ b/avro/avro-symbol-processor/src/main/kotlin/io/koraframework/avro/symbol/processor/AvroSymbolProcessor.kt @@ -0,0 +1,57 @@ +package io.koraframework.avro.symbol.processor + +import com.google.devtools.ksp.processing.CodeGenerator +import com.google.devtools.ksp.processing.Resolver +import com.google.devtools.ksp.processing.SymbolProcessorEnvironment +import com.google.devtools.ksp.symbol.KSAnnotated +import com.google.devtools.ksp.symbol.KSClassDeclaration +import io.koraframework.ksp.common.AnnotationUtils.isAnnotationPresent +import io.koraframework.ksp.common.BaseSymbolProcessor +import io.koraframework.ksp.common.exception.ProcessingErrorException +import io.koraframework.avro.symbol.processor.reader.AvroReaderGenerator +import io.koraframework.avro.symbol.processor.writer.AvroWriterGenerator + +class AvroSymbolProcessor( + environment: SymbolProcessorEnvironment +) : BaseSymbolProcessor(environment) { + private val processed = HashSet() + private val codeGenerator: CodeGenerator = environment.codeGenerator + + private fun getSupportedAnnotationTypes() = setOf( + AvroTypes.avro.canonicalName, + ) + + override fun processRound(resolver: Resolver): List { + val symbolsToProcess = getSupportedAnnotationTypes() + .flatMap { resolver.getSymbolsWithAnnotation(it).toList() } + .distinct() + + val readerGenerator = AvroReaderGenerator(resolver, codeGenerator) + val writerGenerator = AvroWriterGenerator(resolver, codeGenerator) + for (it in symbolsToProcess) { + try { + when (it) { + is KSClassDeclaration -> { + if (it.isAnnotationPresent(AvroTypes.avro)) { + generate(it, readerGenerator, writerGenerator) + } + } + } + } catch (e: ProcessingErrorException) { + e.printError(kspLogger) + } + } + return listOf() + } + + private fun generate( + declaration: KSClassDeclaration, + readerGenerator: AvroReaderGenerator, + writerGenerator: AvroWriterGenerator + ) { + if (processed.add(declaration.qualifiedName!!.asString())) { + readerGenerator.generate(declaration) + writerGenerator.generate(declaration) + } + } +} diff --git a/avro/avro-symbol-processor/src/main/kotlin/io/koraframework/avro/symbol/processor/AvroSymbolProcessorProvider.kt b/avro/avro-symbol-processor/src/main/kotlin/io/koraframework/avro/symbol/processor/AvroSymbolProcessorProvider.kt new file mode 100644 index 000000000..9f41bbef1 --- /dev/null +++ b/avro/avro-symbol-processor/src/main/kotlin/io/koraframework/avro/symbol/processor/AvroSymbolProcessorProvider.kt @@ -0,0 +1,10 @@ +package io.koraframework.avro.symbol.processor + +import com.google.devtools.ksp.processing.SymbolProcessorEnvironment +import com.google.devtools.ksp.processing.SymbolProcessorProvider + +class AvroSymbolProcessorProvider : SymbolProcessorProvider { + override fun create(environment: SymbolProcessorEnvironment): AvroSymbolProcessor { + return AvroSymbolProcessor(environment) + } +} diff --git a/avro/avro-symbol-processor/src/main/kotlin/io/koraframework/avro/symbol/processor/AvroTypes.kt b/avro/avro-symbol-processor/src/main/kotlin/io/koraframework/avro/symbol/processor/AvroTypes.kt new file mode 100644 index 000000000..48ce875a9 --- /dev/null +++ b/avro/avro-symbol-processor/src/main/kotlin/io/koraframework/avro/symbol/processor/AvroTypes.kt @@ -0,0 +1,19 @@ +package io.koraframework.avro.symbol.processor + +import com.squareup.kotlinpoet.ClassName + +object AvroTypes { + + val avro = ClassName("io.koraframework.avro.common.annotation", "Avro") + + val reader = ClassName("io.koraframework.avro.common", "AvroReader") + val writer = ClassName("io.koraframework.avro.common", "AvroWriter") + + val schema = ClassName("org.apache.avro", "Schema") + val specificData = ClassName("org.apache.avro.specific", "SpecificData") + val specificRecord = ClassName("org.apache.avro.specific", "SpecificRecord") + val datumReader = ClassName("org.apache.avro.specific", "SpecificDatumReader") + val datumWriter = ClassName("org.apache.avro.specific", "SpecificDatumWriter") + val decoderFactory = ClassName("org.apache.avro.io", "DecoderFactory") + val encoderFactory = ClassName("org.apache.avro.io", "EncoderFactory") +} diff --git a/avro/avro-symbol-processor/src/main/kotlin/io/koraframework/avro/symbol/processor/JsonUtils.kt b/avro/avro-symbol-processor/src/main/kotlin/io/koraframework/avro/symbol/processor/JsonUtils.kt new file mode 100644 index 000000000..5193cecbe --- /dev/null +++ b/avro/avro-symbol-processor/src/main/kotlin/io/koraframework/avro/symbol/processor/JsonUtils.kt @@ -0,0 +1,10 @@ +package io.koraframework.avro.symbol.processor + +import com.google.devtools.ksp.symbol.KSClassDeclaration +import io.koraframework.ksp.common.generatedClassName + +fun classPackage(classDeclaration: KSClassDeclaration) = classDeclaration.packageName.asString() + +fun KSClassDeclaration.readerName() = this.generatedClassName("AvroReader") + +fun KSClassDeclaration.writerName() = this.generatedClassName("AvroWriter") diff --git a/avro/avro-symbol-processor/src/main/kotlin/io/koraframework/avro/symbol/processor/extension/AvroExtensionFactory.kt b/avro/avro-symbol-processor/src/main/kotlin/io/koraframework/avro/symbol/processor/extension/AvroExtensionFactory.kt new file mode 100644 index 000000000..7bda650a6 --- /dev/null +++ b/avro/avro-symbol-processor/src/main/kotlin/io/koraframework/avro/symbol/processor/extension/AvroExtensionFactory.kt @@ -0,0 +1,21 @@ +package io.koraframework.avro.symbol.processor.extension + +import com.google.devtools.ksp.getClassDeclarationByName +import com.google.devtools.ksp.processing.CodeGenerator +import com.google.devtools.ksp.processing.KSPLogger +import com.google.devtools.ksp.processing.Resolver +import io.koraframework.avro.symbol.processor.AvroTypes +import io.koraframework.kora.app.ksp.extension.ExtensionFactory +import io.koraframework.kora.app.ksp.extension.KoraExtension + +class AvroExtensionFactory : ExtensionFactory { + + override fun create(resolver: Resolver, kspLogger: KSPLogger, codeGenerator: CodeGenerator): KoraExtension? { + val avro = resolver.getClassDeclarationByName(AvroTypes.avro.canonicalName) + return if (avro == null) { + null + } else { + AvroKoraExtension(resolver, kspLogger, codeGenerator) + } + } +} diff --git a/avro/avro-symbol-processor/src/main/kotlin/io/koraframework/avro/symbol/processor/extension/AvroKoraExtension.kt b/avro/avro-symbol-processor/src/main/kotlin/io/koraframework/avro/symbol/processor/extension/AvroKoraExtension.kt new file mode 100644 index 000000000..926bb14aa --- /dev/null +++ b/avro/avro-symbol-processor/src/main/kotlin/io/koraframework/avro/symbol/processor/extension/AvroKoraExtension.kt @@ -0,0 +1,94 @@ +package io.koraframework.avro.symbol.processor.extension + +import com.google.devtools.ksp.getClassDeclarationByName +import com.google.devtools.ksp.processing.CodeGenerator +import com.google.devtools.ksp.processing.KSPLogger +import com.google.devtools.ksp.processing.Resolver +import com.google.devtools.ksp.symbol.KSClassDeclaration +import com.google.devtools.ksp.symbol.KSType +import com.google.devtools.ksp.symbol.Modifier +import com.squareup.kotlinpoet.ClassName +import com.squareup.kotlinpoet.CodeBlock +import io.koraframework.avro.symbol.processor.AvroTypes +import io.koraframework.avro.symbol.processor.reader.AvroReaderGenerator +import io.koraframework.avro.symbol.processor.readerName +import io.koraframework.avro.symbol.processor.writer.AvroWriterGenerator +import io.koraframework.avro.symbol.processor.writerName +import io.koraframework.kora.app.ksp.extension.ExtensionResult +import io.koraframework.kora.app.ksp.extension.ExtensionResult.CodeBlockResult +import io.koraframework.kora.app.ksp.extension.KoraExtension +import io.koraframework.ksp.common.exception.ProcessingErrorException + +class AvroKoraExtension( + private val resolver: Resolver, + kspLogger: KSPLogger, + codeGenerator: CodeGenerator +) : KoraExtension { + private val writerErasure = resolver.getClassDeclarationByName(AvroTypes.writer.canonicalName)!!.asStarProjectedType() + private val readerErasure = resolver.getClassDeclarationByName(AvroTypes.reader.canonicalName)!!.asStarProjectedType() + private val specificRecord = resolver.getClassDeclarationByName(AvroTypes.specificRecord.canonicalName)!!.asStarProjectedType() + private val readerGenerator = AvroReaderGenerator(resolver, codeGenerator) + private val writerGenerator = AvroWriterGenerator(resolver, codeGenerator) + private val generatedMappers = HashSet() + + override fun getDependencyGenerator(resolver: Resolver, type: KSType, tag: String?): (() -> ExtensionResult)? { + if (tag != null && tag != AvroTypes.avro.canonicalName) { + return null + } + + val requestedType = type.makeNotNullable() + if (writerErasure.isAssignableFrom(requestedType)) { + val declaration = dependencyTarget(requestedType, "AvroWriter") ?: return null + return generateMapper(declaration, declaration.writerName(), requestedType, tag) { + writerGenerator.generate(declaration) + } + } + + if (readerErasure.isAssignableFrom(requestedType)) { + val declaration = dependencyTarget(requestedType, "AvroReader") ?: return null + return generateMapper(declaration, declaration.readerName(), requestedType, tag) { + readerGenerator.generate(declaration) + } + } + + return null + } + + private fun dependencyTarget(type: KSType, dependencyName: String): KSClassDeclaration? { + val targetType = type.arguments.firstOrNull()?.type?.resolve() ?: return null + val declaration = targetType.declaration + if (declaration !is KSClassDeclaration + || declaration.modifiers.contains(Modifier.ENUM) + || declaration.modifiers.contains(Modifier.SEALED) + || !specificRecord.isAssignableFrom(targetType) + ) { + throw ProcessingErrorException("$dependencyName can only be created for concrete org.apache.avro.specific.SpecificRecord types", declaration) + } + return declaration + } + + private fun generateMapper( + declaration: KSClassDeclaration, + mapperName: String, + requestedType: KSType, + tag: String?, + generate: () -> Unit + ): () -> ExtensionResult { + return { + val packageName = declaration.packageName.asString() + val mapperCanonicalName = "$packageName.$mapperName" + if (resolver.getClassDeclarationByName(mapperCanonicalName) == null && generatedMappers.add(mapperCanonicalName)) { + generate() + } + + CodeBlockResult( + declaration, + { CodeBlock.of("%T()", ClassName(packageName, mapperName)) }, + requestedType, + tag, + emptyList(), + emptyList() + ) + } + } +} diff --git a/avro/avro-symbol-processor/src/main/kotlin/io/koraframework/avro/symbol/processor/reader/AvroReaderGenerator.kt b/avro/avro-symbol-processor/src/main/kotlin/io/koraframework/avro/symbol/processor/reader/AvroReaderGenerator.kt new file mode 100644 index 000000000..65083748a --- /dev/null +++ b/avro/avro-symbol-processor/src/main/kotlin/io/koraframework/avro/symbol/processor/reader/AvroReaderGenerator.kt @@ -0,0 +1,54 @@ +package io.koraframework.avro.symbol.processor.reader + +import com.google.devtools.ksp.processing.CodeGenerator +import com.google.devtools.ksp.processing.Resolver +import com.google.devtools.ksp.symbol.KSClassDeclaration +import com.squareup.kotlinpoet.* +import com.squareup.kotlinpoet.ParameterizedTypeName.Companion.parameterizedBy +import com.squareup.kotlinpoet.jvm.throws +import com.squareup.kotlinpoet.ksp.addOriginatingKSFile +import com.squareup.kotlinpoet.ksp.toTypeParameterResolver +import com.squareup.kotlinpoet.ksp.toTypeVariableName +import com.squareup.kotlinpoet.ksp.writeTo +import io.koraframework.avro.symbol.processor.AvroTypes +import io.koraframework.avro.symbol.processor.classPackage +import io.koraframework.avro.symbol.processor.readerName +import io.koraframework.ksp.common.KspCommonUtils.generated +import io.koraframework.ksp.common.KspCommonUtils.toTypeName +import java.io.IOException +import java.io.InputStream + +class AvroReaderGenerator(val resolver: Resolver, private val codeGenerator: CodeGenerator) { + fun generate(declaration: KSClassDeclaration) { + val typeName = declaration.toTypeName() + val typeParameterResolver = declaration.typeParameters.toTypeParameterResolver() + val readerInterface = AvroTypes.reader.parameterizedBy(typeName) + val typeBuilder = TypeSpec.classBuilder(declaration.readerName()) + .generated(AvroReaderGenerator::class) + .addAnnotation(AvroTypes.avro) + declaration.containingFile?.let { typeBuilder.addOriginatingKSFile(it) } + + typeBuilder.addSuperinterface(readerInterface) + declaration.typeParameters.forEach { + typeBuilder.addTypeVariable(it.toTypeVariableName(typeParameterResolver)) + } + typeBuilder.addProperty(PropertySpec.builder("SCHEMA", AvroTypes.schema).addModifiers(KModifier.PRIVATE, KModifier.FINAL).initializer("%T.getClassSchema()", typeName).build()) + typeBuilder.addProperty(PropertySpec.builder("SPECIFIC_DATA", AvroTypes.specificData).addModifiers(KModifier.PRIVATE, KModifier.FINAL).initializer("%T().getSpecificData()", typeName).build()) + typeBuilder.addProperty(PropertySpec.builder("READER", AvroTypes.datumReader.parameterizedBy(typeName)).addModifiers(KModifier.PRIVATE, KModifier.FINAL).initializer("%T(SCHEMA, SCHEMA, SPECIFIC_DATA)", AvroTypes.datumReader).build()) + + val method = FunSpec.builder("read") + .addModifiers(KModifier.PUBLIC, KModifier.FINAL, KModifier.OVERRIDE) + .throws(IOException::class) + .addParameter("value", InputStream::class.asTypeName().copy(true)) + .returns(typeName.copy(true)) + method.beginControlFlow("if (value == null || value.available() == 0)") + method.addStatement("return null") + method.endControlFlow() + method.addStatement("val decoder = %T.get().directBinaryDecoder(value, null)", AvroTypes.decoderFactory) + method.addStatement("return READER.read(%T(), decoder)", typeName) + typeBuilder.addFunction(method.build()) + + val spec = typeBuilder.build() + FileSpec.builder(classPackage(declaration), spec.name!!).addType(spec).build().writeTo(codeGenerator = codeGenerator, aggregating = false) + } +} diff --git a/avro/avro-symbol-processor/src/main/kotlin/io/koraframework/avro/symbol/processor/writer/AvroWriterGenerator.kt b/avro/avro-symbol-processor/src/main/kotlin/io/koraframework/avro/symbol/processor/writer/AvroWriterGenerator.kt new file mode 100644 index 000000000..1df9c8d56 --- /dev/null +++ b/avro/avro-symbol-processor/src/main/kotlin/io/koraframework/avro/symbol/processor/writer/AvroWriterGenerator.kt @@ -0,0 +1,57 @@ +package io.koraframework.avro.symbol.processor.writer + +import com.google.devtools.ksp.processing.CodeGenerator +import com.google.devtools.ksp.processing.Resolver +import com.google.devtools.ksp.symbol.KSClassDeclaration +import com.squareup.kotlinpoet.* +import com.squareup.kotlinpoet.ParameterizedTypeName.Companion.parameterizedBy +import com.squareup.kotlinpoet.jvm.throws +import com.squareup.kotlinpoet.ksp.addOriginatingKSFile +import com.squareup.kotlinpoet.ksp.toTypeParameterResolver +import com.squareup.kotlinpoet.ksp.toTypeVariableName +import com.squareup.kotlinpoet.ksp.writeTo +import io.koraframework.avro.symbol.processor.AvroTypes +import io.koraframework.avro.symbol.processor.classPackage +import io.koraframework.avro.symbol.processor.writerName +import io.koraframework.ksp.common.KspCommonUtils.generated +import io.koraframework.ksp.common.KspCommonUtils.toTypeName +import java.io.ByteArrayOutputStream +import java.io.IOException + +class AvroWriterGenerator(val resolver: Resolver, val codeGenerator: CodeGenerator) { + fun generate(declaration: KSClassDeclaration) { + val typeName = declaration.toTypeName() + val typeParameterResolver = declaration.typeParameters.toTypeParameterResolver() + val writerInterface = AvroTypes.writer.parameterizedBy(typeName) + val typeBuilder = TypeSpec.classBuilder(declaration.writerName()) + .generated(AvroWriterGenerator::class) + .addAnnotation(AvroTypes.avro) + declaration.containingFile?.let { typeBuilder.addOriginatingKSFile(it) } + typeBuilder.addSuperinterface(writerInterface) + declaration.typeParameters.forEach { typeBuilder.addTypeVariable(it.toTypeVariableName(typeParameterResolver)) } + + typeBuilder.addProperty(PropertySpec.builder("EMPTY", ByteArray::class).addModifiers(KModifier.PRIVATE, KModifier.FINAL).initializer("%T(0)", ByteArray::class).build()) + typeBuilder.addProperty(PropertySpec.builder("SCHEMA", AvroTypes.schema).addModifiers(KModifier.PRIVATE, KModifier.FINAL).initializer("%T.getClassSchema()", typeName).build()) + typeBuilder.addProperty(PropertySpec.builder("SPECIFIC_DATA", AvroTypes.specificData).addModifiers(KModifier.PRIVATE, KModifier.FINAL).initializer("%T().getSpecificData()", typeName).build()) + typeBuilder.addProperty(PropertySpec.builder("WRITER", AvroTypes.datumWriter.parameterizedBy(typeName)).addModifiers(KModifier.PRIVATE, KModifier.FINAL).initializer("%T(SCHEMA, SPECIFIC_DATA)", AvroTypes.datumWriter).build()) + + val method = FunSpec.builder("writeBytes") + .addModifiers(KModifier.PUBLIC, KModifier.FINAL, KModifier.OVERRIDE) + .throws(IOException::class) + .addParameter("value", typeName.copy(true)) + .returns(ByteArray::class) + method.beginControlFlow("if (value == null)") + method.addStatement("return EMPTY") + method.endControlFlow() + method.beginControlFlow("return %T().%M", ByteArrayOutputStream::class, MemberName("kotlin.io", "use")) + method.addStatement("val encoder = %T.get().directBinaryEncoder(it, null)", AvroTypes.encoderFactory) + method.addStatement("WRITER.write(value, encoder)", typeName) + method.addStatement("encoder.flush()") + method.addStatement("it.toByteArray()") + method.endControlFlow() + typeBuilder.addFunction(method.build()) + + val spec = typeBuilder.build() + FileSpec.builder(classPackage(declaration), spec.name!!).addType(spec).build().writeTo(codeGenerator = codeGenerator, aggregating = false) + } +} diff --git a/avro/avro-symbol-processor/src/main/resources/META-INF/services/com.google.devtools.ksp.processing.SymbolProcessorProvider b/avro/avro-symbol-processor/src/main/resources/META-INF/services/com.google.devtools.ksp.processing.SymbolProcessorProvider new file mode 100644 index 000000000..8b9533f8d --- /dev/null +++ b/avro/avro-symbol-processor/src/main/resources/META-INF/services/com.google.devtools.ksp.processing.SymbolProcessorProvider @@ -0,0 +1 @@ +io.koraframework.avro.symbol.processor.AvroSymbolProcessorProvider diff --git a/avro/avro-symbol-processor/src/main/resources/META-INF/services/io.koraframework.kora.app.ksp.extension.ExtensionFactory b/avro/avro-symbol-processor/src/main/resources/META-INF/services/io.koraframework.kora.app.ksp.extension.ExtensionFactory new file mode 100644 index 000000000..d5f8780c1 --- /dev/null +++ b/avro/avro-symbol-processor/src/main/resources/META-INF/services/io.koraframework.kora.app.ksp.extension.ExtensionFactory @@ -0,0 +1 @@ +io.koraframework.avro.symbol.processor.extension.AvroExtensionFactory diff --git a/avro/avro-symbol-processor/src/test/kotlin/io/koraframework/avro/symbol/processor/AbstractAvroSymbolProcessorTest.kt b/avro/avro-symbol-processor/src/test/kotlin/io/koraframework/avro/symbol/processor/AbstractAvroSymbolProcessorTest.kt new file mode 100644 index 000000000..a72df9152 --- /dev/null +++ b/avro/avro-symbol-processor/src/test/kotlin/io/koraframework/avro/symbol/processor/AbstractAvroSymbolProcessorTest.kt @@ -0,0 +1,150 @@ +package io.koraframework.avro.symbol.processor + +import io.koraframework.avro.common.AvroReader +import io.koraframework.avro.common.AvroWriter +import io.koraframework.ksp.common.AbstractSymbolProcessorTest +import org.apache.avro.generic.IndexedRecord +import org.apache.avro.io.DecoderFactory +import org.apache.avro.io.Encoder +import org.apache.avro.io.EncoderFactory +import org.apache.avro.specific.SpecificData +import org.apache.avro.specific.SpecificDatumReader +import org.apache.avro.specific.SpecificDatumWriter +import org.assertj.core.api.Assertions +import java.io.ByteArrayOutputStream +import java.io.InputStream +import java.nio.charset.StandardCharsets +import java.time.Instant +import java.util.Base64 + +abstract class AbstractAvroSymbolProcessorTest : AbstractSymbolProcessorTest() { + + override fun commonImports(): String { + return super.commonImports() + """ + import io.koraframework.common.annotation.KoraApp; + import iokoraframework.kora.avro.TestAvro; + import io.koraframework.avro.common.annotation.*; + import io.koraframework.avro.common.AvroReader; + import io.koraframework.avro.common.AvroWriter; + """.trimIndent() + } + + protected fun getTestAvroGeneratedRecord(): IndexedRecord { + val testAvro = newGenerated("TestAvro").invoke() as IndexedRecord + testAvro.put(0, "cluster") + testAvro.put(1, Instant.EPOCH) + testAvro.put(2, "descr") + testAvro.put(3, 12345L) + testAvro.put(4, true) + return testAvro + } + + protected fun getTestAvroAsBytes(): ByteArray { + return Base64.getDecoder().decode("DmNsdXN0ZXICAAIKZGVzY3IC8sABAgE=") + } + + protected fun assertThatTestAvroValid(expected: IndexedRecord, actual: IndexedRecord) { + Assertions.assertThat(actual).isNotNull() + Assertions.assertThat(actual[0].toString()).isEqualTo(expected[0].toString()) + Assertions.assertThat(actual[1]).isEqualTo(expected[1]) + Assertions.assertThat(actual[2].toString()).isEqualTo(expected[2].toString()) + Assertions.assertThat(actual[3]).isEqualTo(expected[3]) + Assertions.assertThat(actual[4]).isEqualTo(expected[4]) + } + + protected open fun readerClass(forClass: String) = compileResult.assertSuccess().classLoader.readerClass(testPackage(), forClass) + protected open fun writerClass(forClass: String) = compileResult.assertSuccess().classLoader.writerClass(testPackage(), forClass) + + protected open fun reader(forClass: String, vararg params: Any?) = + compileResult.assertSuccess().classLoader.reader(testPackage(), forClass, *params) + + protected open fun writer(forClass: String, vararg params: Any?) = + compileResult.assertSuccess().classLoader.writer(testPackage(), forClass, *params) + + protected open fun mapperClass(forClass: String) = + compileResult.assertSuccess().classLoader.mapper(testPackage(), forClass) + + protected open fun mapper(forClass: String, readerParams: List<*>, writerParams: List<*>) = + compileResult.assertSuccess().classLoader.mapper(testPackage(), forClass, readerParams, writerParams) + + class ReaderAndWriter(private val reader: AvroReader, private val writer: AvroWriter) : AvroReader, AvroWriter { + + override fun read(`is`: InputStream?): T? = reader.read(`is`) + + override fun writeBytes(value: T?): ByteArray? = writer.writeBytes(value) + } + + companion object { + fun ClassLoader.mapper(packageName: String, forClass: String): ReaderAndWriter { + return mapper(packageName, forClass, listOf(), listOf()) + } + + fun ClassLoader.mapper(packageName: String, forClass: String, readerParams: List<*>, writerParams: List<*>): ReaderAndWriter { + val reader: AvroReader = reader(packageName, forClass, *readerParams.toTypedArray()) + val writer: AvroWriter = writer(packageName, forClass, *writerParams.toTypedArray()) + return ReaderAndWriter(reader, writer) + } + + fun ClassLoader.readerClass(packageName: String, forClass: String) = loadClass(packageName + ".$" + forClass + "_AvroReader")!! + + fun ClassLoader.writerClass(packageName: String, forClass: String) = loadClass(packageName + ".$" + forClass + "_AvroWriter")!! + + fun ClassLoader.reader(packageName: String, forClass: String, vararg params: Any?): AvroReader { + return readerClass(packageName, forClass) + .constructors[0] + .newInstance(*params) as AvroReader + } + + fun ClassLoader.writer(packageName: String, forClass: String, vararg params: Any?): AvroWriter { + return writerClass(packageName, forClass) + .constructors[0] + .newInstance(*params) as AvroWriter + } + + fun ReaderAndWriter.assert(value: T, avro: String) { + this.assertWrite(value, avro) + this.assertRead(avro, value) + } + + fun AvroWriter.assertWrite(value: T, expectedAvro: String) { + Assertions.assertThat(this.writeBytes(value)).asString(StandardCharsets.UTF_8).isEqualTo(expectedAvro) + } + + fun AvroReader.assertRead(avro: String, expectedObject: T) { + Assertions.assertThat(this.read(avro.toByteArray())).isEqualTo(expectedObject) + } + } + + protected fun writeAsBinary(value: IndexedRecord): ByteArray { + try { + ByteArrayOutputStream().use { stream -> + val fieldData = value.javaClass.getDeclaredField("MODEL$") + fieldData.isAccessible = true + val data = fieldData[value] as SpecificData + + val writer = SpecificDatumWriter(value.schema, data) + val encoder: Encoder = EncoderFactory.get().directBinaryEncoder(stream, null) + writer.write(value, encoder) + encoder.flush() + return stream.toByteArray() + } + } catch (e: Exception) { + throw IllegalStateException(e) + } + } + + protected fun readAsBinary(value: ByteArray?): IndexedRecord { + try { + val generated = getTestAvroGeneratedRecord() + val fieldData = generated.javaClass.getDeclaredField("MODEL$") + fieldData.isAccessible = true + val data = fieldData[generated] as SpecificData + + val reader = SpecificDatumReader(generated.schema, generated.schema, data) + val binaryDecoder = DecoderFactory.get().binaryDecoder(value, null) + return reader.read(null, binaryDecoder) as IndexedRecord + } catch (e: Exception) { + throw IllegalStateException(e) + } + } +} diff --git a/avro/avro-symbol-processor/src/test/kotlin/io/koraframework/avro/symbol/processor/AvroTests.kt b/avro/avro-symbol-processor/src/test/kotlin/io/koraframework/avro/symbol/processor/AvroTests.kt new file mode 100644 index 000000000..cf54c6687 --- /dev/null +++ b/avro/avro-symbol-processor/src/test/kotlin/io/koraframework/avro/symbol/processor/AvroTests.kt @@ -0,0 +1,44 @@ +package io.koraframework.avro.symbol.processor + +import iokoraframework.kora.avro.TestAvro +import org.apache.avro.generic.IndexedRecord +import org.assertj.core.api.Assertions.assertThat +import org.junit.jupiter.api.Test +import io.koraframework.avro.common.AvroReader +import io.koraframework.avro.common.AvroWriter +import io.koraframework.kora.app.ksp.KoraAppProcessorProvider +import java.time.Instant + +class AvroTests : AbstractAvroSymbolProcessorTest() { + + @Test + fun testReaderAndWriterFromExtension() { + compile0( + listOf(KoraAppProcessorProvider(), AvroSymbolProcessorProvider()), + """ + @KoraApp + interface TestApp { + @Root + fun root(@Avro r: AvroReader, @Avro w: AvroWriter) = "" + } + """.trimIndent() + ) + + compileResult.assertSuccess() + val reader: AvroReader = compileResult.assertSuccess().classLoader.reader("iokoraframework.kora.avro", "TestAvro") + val writer: AvroWriter = compileResult.assertSuccess().classLoader.writer("iokoraframework.kora.avro", "TestAvro") + assertThat(reader).isNotNull() + assertThat(writer).isNotNull() + + val testAvro: IndexedRecord = TestAvro.newBuilder() + .setCluster("cluster") + .setDate(Instant.EPOCH) + .setDescription("descr") + .setCounter(12345L) + .setFlag(true) + .build() + val write = writer.writeBytes(testAvro) + val read = reader.readUnchecked(write) + assertThatTestAvroValid(testAvro, read!!) + } +} diff --git a/avro/avro-symbol-processor/src/test/resources/avro/TestAvro.avsc b/avro/avro-symbol-processor/src/test/resources/avro/TestAvro.avsc new file mode 100644 index 000000000..4c90c022b --- /dev/null +++ b/avro/avro-symbol-processor/src/test/resources/avro/TestAvro.avsc @@ -0,0 +1,43 @@ +{ + "type": "record", + "name": "TestAvro", + "namespace": "iokoraframework.kora.avro", + "fields": [ + { + "name": "cluster", + "type": "string" + }, + { + "name": "date", + "type": [ + "null", + { + "type": "long", + "logicalType": "timestamp-millis" + } + ] + }, + { + "name": "description", + "type": [ + "null", + "string" + ] + }, + { + "name": "counter", + "type": [ + "null", + "long" + ] + }, + { + "name": "flag", + "type": [ + "null", + "boolean" + ], + "default": null + } + ] +} diff --git a/build.gradle b/build.gradle index e725a1af0..3c2096f60 100644 --- a/build.gradle +++ b/build.gradle @@ -18,6 +18,13 @@ allprojects { version = System.getenv().getOrDefault("KORA_VERSION", "2.0.0-SNAPSHOT") repositories { mavenCentral() + maven { url = "https://packages.confluent.io/maven/" } + } + + buildscript { + repositories { + mavenCentral() + } } } diff --git a/core/kora-app-symbol-processor/src/main/kotlin/io/koraframework/kora/app/ksp/extension/ExtensionResult.kt b/core/kora-app-symbol-processor/src/main/kotlin/io/koraframework/kora/app/ksp/extension/ExtensionResult.kt index 75ce71d49..a78a49557 100644 --- a/core/kora-app-symbol-processor/src/main/kotlin/io/koraframework/kora/app/ksp/extension/ExtensionResult.kt +++ b/core/kora-app-symbol-processor/src/main/kotlin/io/koraframework/kora/app/ksp/extension/ExtensionResult.kt @@ -4,7 +4,11 @@ import com.google.devtools.ksp.symbol.* import com.squareup.kotlinpoet.CodeBlock sealed interface ExtensionResult { - class GeneratedResult(val constructor: KSFunctionDeclaration, val type: KSFunction) : ExtensionResult + + class GeneratedResult( + val constructor: KSFunctionDeclaration, + val type: KSFunction + ) : ExtensionResult class CodeBlockResult( val source: KSDeclaration, @@ -13,8 +17,7 @@ sealed interface ExtensionResult { val componentTag: String?, val dependencyTypes: List, val dependencyTags: List - ) : ExtensionResult { - } + ) : ExtensionResult companion object { fun fromConstructor(constructor: KSFunctionDeclaration, type: KSClassDeclaration): ExtensionResult { diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index 474333501..e1c2c9f27 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -30,6 +30,8 @@ resteasy = "6.2.16.Final" # Data and messaging kafka = "4.3.1" +avro = "1.12.0" +kafka-avro-serializer = "7.7.2" cassandra = "4.19.3" flyway = "12.11.0" @@ -186,6 +188,10 @@ mockserver-netty = { module = "org.mock-server:mockserver-netty", version.ref = mockserver-client = { module = "org.mock-server:mockserver-client-java", version.ref = "mockserver" } kafka-client = { module = "org.apache.kafka:kafka-clients", version.ref = "kafka" } +avro = { module = "org.apache.avro:avro", version.ref = "avro" } +avro-tools = { module = "org.apache.avro:avro-tools", version.ref = "avro" } +kafka-avro-serializer = { module = "io.confluent:kafka-avro-serializer", version.ref = "kafka-avro-serializer" } + # gRPC - https://github.com/grpc/grpc-java/blob/master/SECURITY.md#netty grpc-kotlin-stub = { module = "io.grpc:grpc-kotlin-stub", version.ref = "grpc-kotlin" } diff --git a/kafka/kafka-schema-registry/build.gradle b/kafka/kafka-schema-registry/build.gradle new file mode 100644 index 000000000..5d0268eb1 --- /dev/null +++ b/kafka/kafka-schema-registry/build.gradle @@ -0,0 +1,8 @@ +dependencies { + annotationProcessor project(":config:config-annotation-processor") + + api project(":core:common") + api project(":config:config-common") + + api libs.kafka.avro.serializer +} diff --git a/kafka/kafka-schema-registry/src/main/java/io/koraframework/kafka/schemaregistry/KafkaSchemaRegistryConfig.java b/kafka/kafka-schema-registry/src/main/java/io/koraframework/kafka/schemaregistry/KafkaSchemaRegistryConfig.java new file mode 100644 index 000000000..1d1596caa --- /dev/null +++ b/kafka/kafka-schema-registry/src/main/java/io/koraframework/kafka/schemaregistry/KafkaSchemaRegistryConfig.java @@ -0,0 +1,20 @@ +package io.koraframework.kafka.schemaregistry; + +import io.koraframework.config.common.annotation.ConfigMapper; + +import java.util.List; +import java.util.Map; + +@ConfigMapper +public interface KafkaSchemaRegistryConfig { + + List urls(); + + default int identityMapCapacity() { + return 1000; + } + + default Map properties() { + return Map.of(); + } +} diff --git a/kafka/kafka-schema-registry/src/main/java/io/koraframework/kafka/schemaregistry/KafkaSchemaRegistryModule.java b/kafka/kafka-schema-registry/src/main/java/io/koraframework/kafka/schemaregistry/KafkaSchemaRegistryModule.java new file mode 100644 index 000000000..659bbcbd6 --- /dev/null +++ b/kafka/kafka-schema-registry/src/main/java/io/koraframework/kafka/schemaregistry/KafkaSchemaRegistryModule.java @@ -0,0 +1,19 @@ +package io.koraframework.kafka.schemaregistry; + +import io.confluent.kafka.schemaregistry.client.CachedSchemaRegistryClient; +import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient; +import io.koraframework.common.annotation.DefaultComponent; +import io.koraframework.config.common.Config; +import io.koraframework.config.common.mapper.ConfigValueMapper; + +public interface KafkaSchemaRegistryModule { + + default KafkaSchemaRegistryConfig kafkaSchemaRegistryConfig(Config config, ConfigValueMapper mapper) { + return mapper.mapOrThrow(config.get("kafka.schemaRegistry")); + } + + @DefaultComponent + default SchemaRegistryClient kafkaSchemaRegistryClient(KafkaSchemaRegistryConfig config) { + return new CachedSchemaRegistryClient(config.urls(), config.identityMapCapacity(), config.properties()); + } +} diff --git a/kafka/kafka/build.gradle b/kafka/kafka/build.gradle index 6d8c57880..3ddae0d3e 100644 --- a/kafka/kafka/build.gradle +++ b/kafka/kafka/build.gradle @@ -1,12 +1,20 @@ apply from: "${project.rootDir}/gradle/kotlin-plugin.gradle" dependencies { + annotationProcessor project(":config:config-annotation-processor") + + compileOnly project(":avro:avro-common") + + compileOnly(libs.kafka.avro.serializer) { + exclude group: "org.apache.kafka", module: "kafka-clients" + } + api project(":core:common") api project(":config:config-common") api project(":telemetry:telemetry-common") api project(":logging:logging-common") + api libs.kafka.client - annotationProcessor project(":config:config-annotation-processor") testImplementation project(':internal:test-kafka') } diff --git a/kafka/kafka/src/main/java/io/koraframework/kafka/common/KafkaDeserializersModule.java b/kafka/kafka/src/main/java/io/koraframework/kafka/common/KafkaDeserializersModule.java index e7b2d2416..235b8d99e 100644 --- a/kafka/kafka/src/main/java/io/koraframework/kafka/common/KafkaDeserializersModule.java +++ b/kafka/kafka/src/main/java/io/koraframework/kafka/common/KafkaDeserializersModule.java @@ -2,9 +2,13 @@ import org.apache.kafka.common.serialization.*; import org.apache.kafka.common.utils.Bytes; +import org.apache.avro.specific.SpecificRecord; +import io.koraframework.avro.common.AvroReader; +import io.koraframework.avro.common.annotation.Avro; import io.koraframework.common.annotation.DefaultComponent; import io.koraframework.json.common.JsonReader; import io.koraframework.json.common.annotation.Json; +import io.koraframework.kafka.common.consumer.deserializer.KafkaAvroTypedDeserializer; import io.koraframework.kafka.common.consumer.deserializer.JsonKafkaDeserializer; import java.nio.ByteBuffer; @@ -75,4 +79,10 @@ default Deserializer voidKafkaDeserializer() { default JsonKafkaDeserializer jsonKafkaDeserializer(JsonReader reader) { return new JsonKafkaDeserializer<>(reader); } + + @Avro + @DefaultComponent + default Deserializer avroKafkaSpecificDeserializer(@Avro AvroReader reader) { + return new KafkaAvroTypedDeserializer<>(reader); + } } diff --git a/kafka/kafka/src/main/java/io/koraframework/kafka/common/KafkaSerializersModule.java b/kafka/kafka/src/main/java/io/koraframework/kafka/common/KafkaSerializersModule.java index 6d97ef8c4..8e945e083 100644 --- a/kafka/kafka/src/main/java/io/koraframework/kafka/common/KafkaSerializersModule.java +++ b/kafka/kafka/src/main/java/io/koraframework/kafka/common/KafkaSerializersModule.java @@ -1,10 +1,15 @@ package io.koraframework.kafka.common; +import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient; import org.apache.kafka.common.serialization.*; import org.apache.kafka.common.utils.Bytes; +import org.apache.avro.specific.SpecificRecord; +import io.koraframework.avro.common.AvroWriter; +import io.koraframework.avro.common.annotation.Avro; import io.koraframework.common.annotation.DefaultComponent; import io.koraframework.json.common.JsonWriter; import io.koraframework.json.common.annotation.Json; +import io.koraframework.kafka.common.producer.serializer.KafkaAvroTypedSerializer; import io.koraframework.kafka.common.producer.serializer.JsonKafkaSerializer; import java.nio.ByteBuffer; @@ -75,4 +80,10 @@ default Serializer voidKafkaSerializer() { default Serializer jsonKafkaSerializer(JsonWriter writer) { return new JsonKafkaSerializer<>(writer); } + + @Avro + @DefaultComponent + default Serializer avroKafkaSpecificSerializer(@Avro AvroWriter writer, SchemaRegistryClient schemaRegistry) { + return new KafkaAvroTypedSerializer<>(writer, schemaRegistry); + } } diff --git a/kafka/kafka/src/main/java/io/koraframework/kafka/common/consumer/deserializer/KafkaAvroTypedDeserializer.java b/kafka/kafka/src/main/java/io/koraframework/kafka/common/consumer/deserializer/KafkaAvroTypedDeserializer.java new file mode 100644 index 000000000..b8ef490cd --- /dev/null +++ b/kafka/kafka/src/main/java/io/koraframework/kafka/common/consumer/deserializer/KafkaAvroTypedDeserializer.java @@ -0,0 +1,58 @@ +package io.koraframework.kafka.common.consumer.deserializer; + +import io.koraframework.avro.common.AvroReader; +import org.apache.avro.specific.SpecificRecord; +import org.apache.kafka.common.errors.SerializationException; +import org.apache.kafka.common.header.Headers; +import org.apache.kafka.common.serialization.Deserializer; + +import java.io.IOException; +import java.nio.ByteBuffer; + +public class KafkaAvroTypedDeserializer implements Deserializer { + + protected static final byte MAGIC_BYTE = 0x0; + protected static final int ID_SIZE = 4; + + private final AvroReader avroReader; + + public KafkaAvroTypedDeserializer(AvroReader avroReader) { + this.avroReader = avroReader; + } + + @Override + public T deserialize(String topic, byte[] bytes) { + return deserialize(bytes); + } + + @Override + public T deserialize(String topic, Headers headers, byte[] bytes) { + return deserialize(bytes); + } + + protected T deserialize(byte[] payload) throws SerializationException { + if (payload == null || payload.length == 0) { + return null; + } + + return read(payload); + } + + private T read(byte[] payload) { + if (payload.length <= 1 + ID_SIZE) { + throw new SerializationException("Invalid Avro payload: too short"); + } + if (payload[0] != MAGIC_BYTE) { + throw new SerializationException("Invalid Avro payload: unknown magic byte " + payload[0]); + } + + int schemaId = ByteBuffer.wrap(payload, 1, ID_SIZE).getInt(); + try { + int offset = 1 + ID_SIZE; + ByteBuffer buffer = ByteBuffer.wrap(payload, offset, payload.length - offset); + return avroReader.read(buffer); + } catch (IOException ex) { + throw new SerializationException("Error deserializing Avro message for id " + schemaId, ex); + } + } +} diff --git a/kafka/kafka/src/main/java/io/koraframework/kafka/common/producer/serializer/KafkaAvroTypedSerializer.java b/kafka/kafka/src/main/java/io/koraframework/kafka/common/producer/serializer/KafkaAvroTypedSerializer.java new file mode 100644 index 000000000..215566bc3 --- /dev/null +++ b/kafka/kafka/src/main/java/io/koraframework/kafka/common/producer/serializer/KafkaAvroTypedSerializer.java @@ -0,0 +1,82 @@ +package io.koraframework.kafka.common.producer.serializer; + +import io.confluent.kafka.schemaregistry.avro.AvroSchema; +import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient; +import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException; +import io.koraframework.avro.common.AvroWriter; +import org.apache.avro.specific.SpecificRecord; +import org.apache.kafka.common.errors.SerializationException; +import org.apache.kafka.common.header.Headers; +import org.apache.kafka.common.serialization.Serializer; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.Map; +import java.util.Objects; + +public class KafkaAvroTypedSerializer implements Serializer { + + protected static final byte MAGIC_BYTE = 0x0; + protected static final int ID_SIZE = 4; + private static final String AUTO_REGISTER_SCHEMAS_CONFIG = "auto.register.schemas"; + + private final AvroWriter avroWriter; + private final SchemaRegistryClient schemaRegistry; + private boolean autoRegisterSchemas = true; + private boolean isKey; + + public KafkaAvroTypedSerializer(AvroWriter avroWriter, SchemaRegistryClient schemaRegistry) { + this.avroWriter = Objects.requireNonNull(avroWriter); + this.schemaRegistry = Objects.requireNonNull(schemaRegistry); + } + + @Override + public void configure(Map configs, boolean isKey) { + this.isKey = isKey; + var autoRegisterSchemas = configs.get(AUTO_REGISTER_SCHEMAS_CONFIG); + this.autoRegisterSchemas = autoRegisterSchemas == null || Boolean.parseBoolean(String.valueOf(autoRegisterSchemas)); + } + + @Override + public byte[] serialize(String topic, T data) { + return this.serialize(topic, null, data); + } + + @Override + public byte[] serialize(String topic, Headers headers, T record) { + if (record == null) { + return null; + } + + AvroSchema schema = new AvroSchema(record.getSchema()); + return serializeImpl(topic, record, schema); + } + + protected byte[] serializeImpl(String topic, T object, AvroSchema schema) throws SerializationException { + if (object == null) { + return null; + } + + try { + int id = schemaId(topic, schema); + try (ByteArrayOutputStream out = new ByteArrayOutputStream()) { + out.write(MAGIC_BYTE); + out.write(ByteBuffer.allocate(ID_SIZE).putInt(id).array()); + out.write(avroWriter.writeBytesUnchecked(object)); + return out.toByteArray(); + } + } catch (IOException | RuntimeException e) { + throw new SerializationException("Error serializing Avro message", e); + } catch (RestClientException e) { + throw new SerializationException("Error registering Avro schema", e); + } + } + + private int schemaId(String topic, AvroSchema schema) throws IOException, RestClientException { + var subject = topic + (this.isKey ? "-key" : "-value"); + return this.autoRegisterSchemas + ? this.schemaRegistry.register(subject, schema) + : this.schemaRegistry.getId(subject, schema); + } +} diff --git a/kora-app-symbol-processor/src/main/kotlin/ru/tinkoff/kora/kora/app/ksp/extension/KoraExtension.kt b/kora-app-symbol-processor/src/main/kotlin/ru/tinkoff/kora/kora/app/ksp/extension/KoraExtension.kt new file mode 100644 index 000000000..e69de29bb diff --git a/settings.gradle b/settings.gradle index 5267f2a03..c24cac74e 100644 --- a/settings.gradle +++ b/settings.gradle @@ -52,6 +52,7 @@ include( 'database:database-liquibase', 'jms', 'kafka:kafka', + 'kafka:kafka-schema-registry', 'kafka:kafka-annotation-processor', 'kafka:kafka-symbol-processor', 'http:http-common', @@ -105,4 +106,7 @@ include( 'experimental:camunda-zeebe-worker', 'experimental:camunda-zeebe-worker-annotation-processor', 'experimental:camunda-zeebe-worker-symbol-processor', + 'avro:avro-common', + 'avro:avro-annotation-processor', + 'avro:avro-symbol-processor', )