417 issue cant load an avro schema serializer from register

config/checkstyle/OSS_checkstyle_suppressions.xml

@@ -15,6 +15,7 @@
   <suppress checks="ClassDataAbstractionCoupling" files="[/\\].*ProcessorTest\.java" />
   <suppress checks="ClassDataAbstractionCoupling" files="[/\\].*Standalone\.java" />
   <suppress checks="ClassDataAbstractionCoupling" files="[/\\].*AvroSerializersUtil\.java" />
+  <suppress checks="ClassDataAbstractionCoupling" files="[/\\].*ExtractorFactoryTest\.java" />
   <suppress checks="ExecutableStatementCount" files="[/\\].*Standalone\.java" />
   <suppress checks="AnonInnerLength" files="[/\\].*AutoCompletion\.java" />
   <suppress checks="Indentation" files="[/\\].*ExtractorTest\.java" />

pom-maven-central.xml

@@ -301,6 +301,17 @@
       </roles>
       <timezone>Europe/Madrid</timezone>
     </developer>
+    <developer>
+      <id>RobertoSngular</id>
+      <name>Roberto Riveira Veiga</name>
+      <email>[email protected]</email>
+      <organization>Sngular</organization>
+      <organizationUrl>https://sngular.github.io/</organizationUrl>
+      <roles>
+        <role>Backend Developer</role>
+      </roles>
+      <timezone>Europe/Madrid</timezone>
+    </developer>
   </developers>
 
   <scm>

pom.xml

@@ -7,7 +7,7 @@
 
   <artifactId>kloadgen</artifactId>
 
-  <version>5.6.10</version>
+  <version>5.6.11</version>
 
   <name>KLoadGen</name>
   <description>Load Generation Jmeter plugin for Kafka Cluster. Supporting AVRO, JSON Schema and Protobuf schema types. Generate Artificial
@@ -310,6 +310,27 @@
       <roles>
         <role>Trainee Backend Developer</role>
       </roles>
+    </developer>
+    <developer>
+      <id>GraciMndzSNG</id>
+      <name>Graciela Méndez Olmos</name>
+      <email>[email protected]</email>
+      <organization>Sngular</organization>
+      <organizationUrl>https://sngular.github.io/</organizationUrl>
+      <roles>
+        <role>Backend Developer</role>
+      </roles>
+      <timezone>Europe/Madrid</timezone>
+    </developer>
+    <developer>
+      <id>pablorodriguez-sngular</id>
+      <name>Pablo Rodríguez Pérez</name>
+      <email>[email protected]</email>
+      <organization>Sngular</organization>
+      <organizationUrl>https://sngular.github.io/</organizationUrl>
+      <roles>
+        <role>Backend Developer</role>
+      </roles>
       <timezone>Europe/Madrid</timezone>
     </developer>
   </developers>

src/main/java/com/sngular/kloadgen/config/valuefileserialized/ValueFileSerializedConfigElementBeanInfo.java

@@ -42,33 +42,23 @@ public ValueFileSerializedConfigElementBeanInfo() {
 
     final PropertyDescriptor nameStrategyPropertyProps = property(VALUE_NAME_STRATEGY);
     nameStrategyPropertyProps.setPropertyEditorClass(NameStrategyPropertyEditor.class);
-    nameStrategyPropertyProps.setValue(NOT_UNDEFINED, Boolean.TRUE);
-    nameStrategyPropertyProps.setValue(DEFAULT, "");
-    nameStrategyPropertyProps.setValue(NOT_EXPRESSION, Boolean.FALSE);
+    complete(nameStrategyPropertyProps);
 
     final PropertyDescriptor serializerPropertyProps = property(VALUE_SERIALIZER_PROPERTY);
     serializerPropertyProps.setPropertyEditorClass(ValueSerializerPropertyEditor.class);
-    serializerPropertyProps.setValue(NOT_UNDEFINED, Boolean.TRUE);
-    serializerPropertyProps.setValue(DEFAULT, "");
-    serializerPropertyProps.setValue(NOT_EXPRESSION, Boolean.FALSE);
+    complete(serializerPropertyProps);
 
     final PropertyDescriptor subjectNameProps = property(VALUE_SUBJECT_NAME);
     subjectNameProps.setPropertyEditorClass(FileSubjectPropertyEditor.class);
-    subjectNameProps.setValue(NOT_UNDEFINED, Boolean.TRUE);
-    subjectNameProps.setValue(DEFAULT, "");
-    subjectNameProps.setValue(NOT_EXPRESSION, Boolean.FALSE);
+    complete(subjectNameProps);
 
     final PropertyDescriptor schemaType = property(VALUE_SCHEMA_TYPE);
     schemaType.setPropertyEditorClass(SchemaTypePropertyEditor.class);
-    schemaType.setValue(NOT_UNDEFINED, Boolean.TRUE);
-    schemaType.setValue(DEFAULT, "");
-    schemaType.setValue(NOT_EXPRESSION, Boolean.FALSE);
+    complete(schemaType);
 
     final PropertyDescriptor avroSchemaProps = property(VALUE_SCHEMA_DEFINITION);
     avroSchemaProps.setPropertyEditorClass(SchemaConverterPropertyEditor.class);
-    avroSchemaProps.setValue(NOT_UNDEFINED, Boolean.TRUE);
-    avroSchemaProps.setValue(DEFAULT, "");
-    avroSchemaProps.setValue(NOT_EXPRESSION, Boolean.FALSE);
+    complete(avroSchemaProps);
 
     final TypeEditor tableEditor = TypeEditor.TableEditor;
     final PropertyDescriptor tableProperties = property(VALUE_SCHEMA_PROPERTIES, tableEditor);
@@ -79,4 +69,10 @@ public ValueFileSerializedConfigElementBeanInfo() {
     tableProperties.setValue(DEFAULT, new ArrayList<>());
     tableProperties.setValue(NOT_UNDEFINED, Boolean.TRUE);
   }
+
+  private void complete(final PropertyDescriptor propertyDescriptor) {
+    propertyDescriptor.setValue(NOT_UNDEFINED, Boolean.TRUE);
+    propertyDescriptor.setValue(DEFAULT, "");
+    propertyDescriptor.setValue(NOT_EXPRESSION, Boolean.FALSE);
+  }
 }

src/main/java/com/sngular/kloadgen/extractor/SchemaExtractor.java

@@ -10,7 +10,7 @@
 import com.sngular.kloadgen.common.SchemaRegistryEnum;
 import com.sngular.kloadgen.extractor.extractors.ExtractorFactory;
 import com.sngular.kloadgen.model.FieldValueMapping;
-import io.confluent.kafka.schemaregistry.ParsedSchema;
+import com.sngular.kloadgen.parsedschema.ParsedSchema;
 import lombok.Getter;
 import org.apache.commons.lang3.tuple.Pair;
 

src/main/java/com/sngular/kloadgen/extractor/extractors/ExtractorFactory.java

@@ -10,8 +10,9 @@
 import com.sngular.kloadgen.exception.KLoadGenException;
 import com.sngular.kloadgen.extractor.extractors.avro.AvroExtractor;
 import com.sngular.kloadgen.extractor.extractors.json.JsonExtractor;
-import com.sngular.kloadgen.extractor.extractors.protobuff.ProtobuffExtractor;
+import com.sngular.kloadgen.extractor.extractors.protobuf.ProtobufExtractor;
 import com.sngular.kloadgen.model.FieldValueMapping;
+import com.sngular.kloadgen.parsedschema.ParsedSchema;
 import com.sngular.kloadgen.schemaregistry.adapter.impl.AbstractParsedSchemaAdapter;
 import com.sngular.kloadgen.schemaregistry.adapter.impl.ApicurioAbstractParsedSchemaMetadata;
 import com.sngular.kloadgen.util.JMeterHelper;
@@ -21,22 +22,28 @@
 import org.apache.jmeter.threads.JMeterContextService;
 
 public final class ExtractorFactory {
-  private static final AvroExtractor AVRO_EXTRACTOR = new AvroExtractor();
+  private static AvroExtractor avroExtractor = new AvroExtractor();
 
-  private static final JsonExtractor JSON_EXTRACTOR = new JsonExtractor();
+  private static JsonExtractor jsonExtractor = new JsonExtractor();
 
-  private static final ProtobuffExtractor PROTOBUFF_EXTRACTOR = new ProtobuffExtractor();
+  private static ProtobufExtractor protobufExtractor = new ProtobufExtractor();
 
   private ExtractorFactory() {
   }
 
+  public static void configExtractorFactory(final AvroExtractor avroExtractor, final JsonExtractor jsonExtractor, final ProtobufExtractor protobufExtractor) {
+    ExtractorFactory.avroExtractor = avroExtractor;
+    ExtractorFactory.jsonExtractor = jsonExtractor;
+    ExtractorFactory.protobufExtractor = protobufExtractor;
+  }
+
   public static ExtractorRegistry getExtractor(final String schemaType) {
 
     if (schemaType != null && EnumUtils.isValidEnum(SchemaTypeEnum.class, schemaType.toUpperCase())) {
       final ExtractorRegistry response = switch (SchemaTypeEnum.valueOf(schemaType.toUpperCase())) {
-        case JSON -> JSON_EXTRACTOR;
-        case AVRO -> AVRO_EXTRACTOR;
-        case PROTOBUF -> PROTOBUFF_EXTRACTOR;
+        case JSON -> jsonExtractor;
+        case AVRO -> avroExtractor;
+        case PROTOBUF -> protobufExtractor;
         default -> throw new KLoadGenException(String.format("Schema type not supported %s", schemaType));
       };
       return response;
@@ -62,7 +69,7 @@ public static Pair<String, List<FieldValueMapping>> flatPropertiesList(final Str
         case APICURIO -> ((ApicurioAbstractParsedSchemaMetadata) abstractParsedSchemaAdapter).getSchema();
         case CONFLUENT -> abstractParsedSchemaAdapter.getRawSchema();
       };
-      attributeList.addAll(getExtractor(schemaType).processSchema(schema, schemaRegistryEnum));
+      attributeList.addAll(getExtractor(schemaType).processSchema(new ParsedSchema(schema), schemaRegistryEnum));
     }
     return Pair.of(schemaType, attributeList);
   }

src/main/java/com/sngular/kloadgen/extractor/extractors/avro/AvroConfluentExtractor.java

@@ -4,16 +4,16 @@
 
 import com.sngular.kloadgen.extractor.extractors.Extractor;
 import com.sngular.kloadgen.model.FieldValueMapping;
-import io.confluent.kafka.schemaregistry.avro.AvroSchema;
+import org.apache.avro.Schema;
 
-public class AvroConfluentExtractor extends AbstractAvroFileExtractor implements Extractor<AvroSchema> {
+public class AvroConfluentExtractor extends AbstractAvroFileExtractor implements Extractor<Schema> {
 
-  public final List<FieldValueMapping> processSchema(final AvroSchema schema) {
-    return this.processSchemaDefault(schema.rawSchema());
+  public final List<FieldValueMapping> processSchema(final Schema schema) {
+    return this.processSchemaDefault(schema);
   }
 
   public final List<String> getSchemaNameList(final String schema) {
-    return getSchemaNameList(new AvroSchema(schema).rawSchema());
+    return getSchemaNameList(new Schema.Parser().parse(schema));
   }
 
 }

src/main/java/com/sngular/kloadgen/extractor/extractors/avro/AvroExtractor.java

@@ -7,20 +7,19 @@
 import com.sngular.kloadgen.extractor.extractors.Extractor;
 import com.sngular.kloadgen.extractor.extractors.ExtractorRegistry;
 import com.sngular.kloadgen.model.FieldValueMapping;
-import io.confluent.kafka.schemaregistry.ParsedSchema;
-import io.confluent.kafka.schemaregistry.avro.AvroSchema;
+import com.sngular.kloadgen.parsedschema.ParsedSchema;
 
-public class AvroExtractor implements ExtractorRegistry<Object> {
+public class AvroExtractor implements ExtractorRegistry<ParsedSchema> {
 
   private static Map<SchemaRegistryEnum, Extractor> schemaRegistryMap = Map.of(SchemaRegistryEnum.CONFLUENT, new AvroConfluentExtractor(), SchemaRegistryEnum.APICURIO,
                                                                        new AvroApicurioExtractor());
 
-  public final List<FieldValueMapping> processSchema(final Object schema, final SchemaRegistryEnum registryEnum) {
-    return schemaRegistryMap.get(registryEnum).processSchema(schema);
+  public final List<FieldValueMapping> processSchema(final ParsedSchema schema, final SchemaRegistryEnum registryEnum) {
+    return schemaRegistryMap.get(registryEnum).processSchema(schema.rawSchema());
   }
 
   public final ParsedSchema processSchema(final String fileContent) {
-    return new AvroSchema(fileContent);
+    return new ParsedSchema(fileContent, "AVRO");
   }
 
   public final List<String> getSchemaNameList(final String schema, final SchemaRegistryEnum registryEnum) {

src/main/java/com/sngular/kloadgen/extractor/extractors/json/JsonDefaultExtractor.java

@@ -16,7 +16,7 @@ public class JsonDefaultExtractor extends AbstractJsonExtractor implements Extra
   @Override
   public final List<FieldValueMapping> processSchema(final String schema) {
 
-    final Schema parsed = getSchemaParser().parse(schema.toString());
+    final Schema parsed = getSchemaParser().parse(schema);
 
     final List<FieldValueMapping> attributeList = new ArrayList<>();
     parsed.getProperties().forEach(field -> attributeList.addAll(processField(field, true, null)));

src/main/java/com/sngular/kloadgen/extractor/extractors/json/JsonExtractor.java

@@ -7,21 +7,20 @@
 import com.sngular.kloadgen.extractor.extractors.Extractor;
 import com.sngular.kloadgen.extractor.extractors.ExtractorRegistry;
 import com.sngular.kloadgen.model.FieldValueMapping;
-import io.confluent.kafka.schemaregistry.ParsedSchema;
-import io.confluent.kafka.schemaregistry.json.JsonSchema;
+import com.sngular.kloadgen.parsedschema.ParsedSchema;
 
 
-public class JsonExtractor implements ExtractorRegistry<Object> {
+public class JsonExtractor implements ExtractorRegistry<ParsedSchema> {
 
   private static final Map<SchemaRegistryEnum, Extractor> SCHEMA_REGISTRY_MAP = Map.of(SchemaRegistryEnum.CONFLUENT, new JsonDefaultExtractor(),
                                                                                        SchemaRegistryEnum.APICURIO, new JsonDefaultExtractor());
 
-  public final List<FieldValueMapping> processSchema(final Object schemaReceived, final SchemaRegistryEnum registryEnum) {
-    return SCHEMA_REGISTRY_MAP.get(registryEnum).processSchema(schemaReceived.toString());
+  public final List<FieldValueMapping> processSchema(final ParsedSchema schemaReceived, final SchemaRegistryEnum registryEnum) {
+    return SCHEMA_REGISTRY_MAP.get(registryEnum).processSchema(schemaReceived.rawSchema().toString());
   }
 
   public final ParsedSchema processSchema(final String fileContent) {
-    return new JsonSchema(fileContent);
+    return new ParsedSchema(fileContent, "JSON");
   }
 
   public final List<String> getSchemaNameList(final String schema, final SchemaRegistryEnum registryEnum) {

src/main/java/com/sngular/kloadgen/extractor/extractors/protobuff/AbstractProtoFileExtractor.java → src/main/java/com/sngular/kloadgen/extractor/extractors/protobuf/AbstractProtoFileExtractor.java

@@ -1,4 +1,4 @@
-package com.sngular.kloadgen.extractor.extractors.protobuff;
+package com.sngular.kloadgen.extractor.extractors.protobuf;
 
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;

src/main/java/com/sngular/kloadgen/extractor/extractors/protobuff/ProtoBufApicurioExtractor.java → src/main/java/com/sngular/kloadgen/extractor/extractors/protobuf/ProtoBufApicurioExtractor.java

@@ -1,4 +1,4 @@
-package com.sngular.kloadgen.extractor.extractors.protobuff;
+package com.sngular.kloadgen.extractor.extractors.protobuf;
 
 import java.util.List;
 

src/main/java/com/sngular/kloadgen/extractor/extractors/protobuff/ProtoBufConfluentExtractor.java → src/main/java/com/sngular/kloadgen/extractor/extractors/protobuf/ProtoBufConfluentExtractor.java

@@ -1,15 +1,17 @@
-package com.sngular.kloadgen.extractor.extractors.protobuff;
+package com.sngular.kloadgen.extractor.extractors.protobuf;
 
 import java.util.List;
 
 import com.sngular.kloadgen.extractor.extractors.Extractor;
 import com.sngular.kloadgen.model.FieldValueMapping;
+import com.sngular.kloadgen.parsedschema.ParsedSchema;
+import com.squareup.wire.schema.internal.parser.ProtoFileElement;
 import io.confluent.kafka.schemaregistry.protobuf.ProtobufSchema;
 
-public class ProtoBufConfluentExtractor extends AbstractProtoFileExtractor implements Extractor<ProtobufSchema> {
+public class ProtoBufConfluentExtractor extends AbstractProtoFileExtractor implements Extractor<ParsedSchema> {
 
-  public final List<FieldValueMapping> processSchema(final ProtobufSchema schemaReceived) {
-    return processSchemaDefault(schemaReceived.rawSchema());
+  public final List<FieldValueMapping> processSchema(final ParsedSchema schemaReceived) {
+    return processSchemaDefault((ProtoFileElement) schemaReceived.rawSchema());
   }
 
   public final List<String> getSchemaNameList(final String schema) {

src/main/java/com/sngular/kloadgen/extractor/extractors/protobuff/ProtobuffExtractor.java → src/main/java/com/sngular/kloadgen/extractor/extractors/protobuf/ProtobufExtractor.java

@@ -1,4 +1,4 @@
-package com.sngular.kloadgen.extractor.extractors.protobuff;
+package com.sngular.kloadgen.extractor.extractors.protobuf;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -8,27 +8,19 @@
 import com.sngular.kloadgen.extractor.extractors.Extractor;
 import com.sngular.kloadgen.extractor.extractors.ExtractorRegistry;
 import com.sngular.kloadgen.model.FieldValueMapping;
-import com.squareup.wire.schema.internal.parser.ProtoFileElement;
-import io.confluent.kafka.schemaregistry.ParsedSchema;
-import io.confluent.kafka.schemaregistry.protobuf.ProtobufSchema;
+import com.sngular.kloadgen.parsedschema.ParsedSchema;
 
-public class ProtobuffExtractor implements ExtractorRegistry<Object> {
+public class ProtobufExtractor implements ExtractorRegistry<ParsedSchema> {
 
   private static final Map<SchemaRegistryEnum, Extractor> SCHEMA_REGISTRY_MAP = Map.of(SchemaRegistryEnum.CONFLUENT, new ProtoBufConfluentExtractor(),
                                                                                        SchemaRegistryEnum.APICURIO, new ProtoBufApicurioExtractor());
 
-  public final List<FieldValueMapping> processSchema(final Object schemaReceived, final SchemaRegistryEnum registryEnum) {
-    final var resultSchema = new ArrayList<FieldValueMapping>();
-    if (schemaReceived instanceof ProtoFileElement) {
-      resultSchema.addAll(SCHEMA_REGISTRY_MAP.get(SchemaRegistryEnum.APICURIO).processSchema(schemaReceived));
-    } else {
-      resultSchema.addAll(SCHEMA_REGISTRY_MAP.get(registryEnum).processSchema(schemaReceived));
-    }
-    return resultSchema;
+  public final List<FieldValueMapping> processSchema(final ParsedSchema schemaReceived, final SchemaRegistryEnum registryEnum) {
+    return new ArrayList<FieldValueMapping>(SCHEMA_REGISTRY_MAP.get(registryEnum).processSchema(schemaReceived));
   }
 
   public final ParsedSchema processSchema(final String fileContent) {
-    return new ProtobufSchema(fileContent);
+    return new ParsedSchema(fileContent, "PROTOBUF");
   }
 
   public final List<String> getSchemaNameList(final String schema, final SchemaRegistryEnum registryEnum) {

src/main/java/com/sngular/kloadgen/loadgen/impl/ProtobufLoadGenerator.java

@@ -45,7 +45,7 @@ public final void setUpGenerator(
   public final void setUpGenerator(final String schema, final List<FieldValueMapping> fieldExprMappings) {
     final ProtobufSchema protobufSchema = new ProtobufSchema(schema);
     this.protobufSchemaProcessor
-        .processSchema(SchemaTypeEnum.PROTOBUF, protobufSchema, new BaseSchemaMetadata<>(
+        .processSchema(SchemaTypeEnum.PROTOBUF, protobufSchema.rawSchema(), new BaseSchemaMetadata<>(
                            ConfluentSchemaMetadata.parse(new SchemaMetadata(1, 1, SchemaTypeEnum.PROTOBUF.name(), Collections.emptyList(), schema))),
                        fieldExprMappings);
   }

src/main/java/com/sngular/kloadgen/parsedschema/IParsedSchema.java

@@ -0,0 +1,13 @@
+package com.sngular.kloadgen.parsedschema;
+
+public interface IParsedSchema<T> {
+
+  String schemaType();
+
+  String name();
+
+  String canonicalString();
+
+  Object rawSchema();
+
+}