Skip to content

Commit 5ead042

Browse files
pablorodriguez-sngularpablorodriguez-sngular
authored
422 issue nullpointerexception when starting a test with a key schema file serializer config (#426)
Fixed issues relating to how Serializer classes are used in KafkaProducer. * Fixed issues with missing variables * Removed hardcoded serializer list and changed constructor * Corrected constructor to actually use classes. * Converted EnrichedRecord and StringSerializer to work with each other and other Serializers. * Added missing parameter in constructor * Added cleanup of properties * Modified how key or value is detected. * Removed some changes from 68ce277, switched to a cherry-pick from 661f3b4. * Fixed issue about wrong value serializer when switching from ValueFileSerializer to KeyFileSerializer. * Changed how we choose the Record for KafkaProducerSampler * Checkstyle fixes
1 parent 69aab7e commit 5ead042

File tree

7 files changed

+59
-62
lines changed

7 files changed

+59
-62
lines changed

‎src/main/java/com/sngular/kloadgen/config/schemaregistry/SchemaRegistryConfigElementBeanInfo.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,8 @@ public SchemaRegistryConfigElementBeanInfo() {
2323
super(SchemaRegistryConfigElement.class);
2424

2525
createPropertyGroup("schema_registry_config", new String[] {SchemaRegistryConfigElementValue.SCHEMA_REGISTRY_NAME,
26-
SchemaRegistryConfigElementValue.SCHEMA_REGISTRY_URL, SchemaRegistryConfigElementValue.SCHEMA_REGISTRY_PROPERTIES});
26+
SchemaRegistryConfigElementValue.SCHEMA_REGISTRY_URL,
27+
SchemaRegistryConfigElementValue.SCHEMA_REGISTRY_PROPERTIES});
2728

2829
final PropertyDescriptor schemaRegistryName = property(SchemaRegistryConfigElementValue.SCHEMA_REGISTRY_NAME);
2930
schemaRegistryName.setPropertyEditorClass(SchemaRegistryNamePropertyEditor.class);

‎src/main/java/com/sngular/kloadgen/property/editor/FileSubjectPropertyEditor.java‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -179,5 +179,4 @@ public final Component getCustomEditor() {
179179
public final boolean supportsCustomEditor() {
180180
return true;
181181
}
182-
183182
}

‎src/main/java/com/sngular/kloadgen/property/editor/ReflectionUtils.java‎

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -17,12 +17,12 @@
1717
import org.apache.kafka.common.serialization.Serializer;
1818
import org.reflections.Reflections;
1919

20-
final class ReflectionUtils {
20+
public final class ReflectionUtils {
2121

2222
private ReflectionUtils() {
2323
}
2424

25-
static void extractSerializers(final JComboBox<String> serializerComboBox, final Reflections reflections, final Class reflectedClass) {
25+
public static List<String> extractSerializers(final Reflections reflections, final Class reflectedClass) {
2626
final Set<Class<? extends Serializer>> subTypes = reflections.getSubTypesOf(reflectedClass);
2727
final List<String> classList = new ArrayList<>();
2828

@@ -31,7 +31,11 @@ static void extractSerializers(final JComboBox<String> serializerComboBox, final
3131
}
3232

3333
classList.sort(Comparator.naturalOrder());
34-
for (String serializer : classList) {
34+
return classList;
35+
}
36+
37+
static void extractSerializers(final JComboBox<String> serializerComboBox, final Reflections reflections, final Class reflectedClass) {
38+
for (String serializer : extractSerializers(reflections, reflectedClass)) {
3539
serializerComboBox.addItem(serializer);
3640
}
3741
serializerComboBox.setSelectedItem(0);

‎src/main/java/com/sngular/kloadgen/property/editor/SchemaConverterPropertyEditor.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -48,7 +48,8 @@ public final void actionPerformed(final ActionEvent event) {
4848
}
4949

5050
@Override
51-
public final void clearGui() {}
51+
public final void clearGui() {
52+
}
5253

5354
@Override
5455
public final void setDescriptor(final PropertyDescriptor descriptor) {

‎src/main/java/com/sngular/kloadgen/sampler/KafkaProducerSampler.java‎

Lines changed: 45 additions & 47 deletions
Original file line numberDiff line numberDiff line change
@@ -8,27 +8,24 @@
88

99
import java.io.Serial;
1010
import java.io.Serializable;
11+
import java.lang.reflect.InvocationTargetException;
1112
import java.nio.charset.StandardCharsets;
1213
import java.util.ArrayList;
1314
import java.util.Collections;
1415
import java.util.List;
1516
import java.util.Objects;
1617
import java.util.Properties;
17-
import java.util.Set;
1818
import java.util.concurrent.ExecutionException;
1919

2020
import com.sngular.kloadgen.exception.KLoadGenException;
2121
import com.sngular.kloadgen.loadgen.BaseLoadGenerator;
2222
import com.sngular.kloadgen.model.HeaderMapping;
2323
import com.sngular.kloadgen.randomtool.generator.StatelessGeneratorTool;
24-
import com.sngular.kloadgen.serializer.AvroSerializer;
2524
import com.sngular.kloadgen.serializer.EnrichedRecord;
26-
import com.sngular.kloadgen.serializer.ProtobufSerializer;
2725
import com.sngular.kloadgen.util.ProducerKeysHelper;
2826
import com.sngular.kloadgen.util.PropsKeysHelper;
2927
import io.apicurio.registry.serde.Legacy4ByteIdHandler;
3028
import io.apicurio.registry.serde.SerdeConfig;
31-
import io.apicurio.registry.serde.avro.AvroKafkaSerializer;
3229
import org.apache.commons.lang3.StringUtils;
3330
import org.apache.jmeter.config.Arguments;
3431
import org.apache.jmeter.protocol.java.sampler.AbstractJavaSamplerClient;
@@ -41,14 +38,12 @@
4138
import org.apache.kafka.clients.producer.ProducerRecord;
4239
import org.apache.kafka.clients.producer.RecordMetadata;
4340
import org.apache.kafka.common.KafkaException;
44-
import org.apache.kafka.common.serialization.StringSerializer;
41+
import org.apache.kafka.common.serialization.Serializer;
4542

4643
public final class KafkaProducerSampler extends AbstractJavaSamplerClient implements Serializable {
4744

4845
private static final String TEMPLATE = "Topic: %s, partition: %s, offset: %s";
4946

50-
private static final Set<String> SERIALIZER_SET = Set.of(AvroSerializer.class.getName(), ProtobufSerializer.class.getName());
51-
5247
@Serial
5348
private static final long serialVersionUID = 1L;
5449

@@ -73,42 +68,49 @@ public final class KafkaProducerSampler extends AbstractJavaSamplerClient implem
7368
@Override
7469
public void setupTest(final JavaSamplerContext context) {
7570
props = JMeterContextService.getContext().getProperties();
76-
try {
71+
72+
if (context.getJMeterVariables().get(PropsKeysHelper.VALUE_SCHEMA) == null) {
73+
generator = SamplerUtil.configureKeyGenerator(props);
74+
} else {
7775
generator = SamplerUtil.configureValueGenerator(props);
76+
}
7877

79-
if ("true".equals(JavaSamplerContext.getJMeterVariables().get(PropsKeysHelper.SCHEMA_KEYED_MESSAGE_KEY))
80-
|| "true".equals(JavaSamplerContext.getJMeterVariables().get(PropsKeysHelper.SIMPLE_KEYED_MESSAGE_KEY))) {
81-
keyMessageFlag = true;
82-
if (!Objects.isNull(JMeterContextService.getContext().getVariables().get(PropsKeysHelper.KEY_SUBJECT_NAME))) {
83-
keyGenerator = SamplerUtil.configureKeyGenerator(props);
84-
} else {
85-
msgKeyType = props.getProperty(PropsKeysHelper.MESSAGE_KEY_KEY_TYPE);
86-
msgKeyValue = PropsKeysHelper.MSG_KEY_VALUE.equalsIgnoreCase(props.getProperty(PropsKeysHelper.MESSAGE_KEY_KEY_VALUE))
87-
? Collections.emptyList() : Collections.singletonList(props.getProperty(PropsKeysHelper.MESSAGE_KEY_KEY_VALUE));
88-
}
78+
if ("true".equals(context.getJMeterVariables().get(PropsKeysHelper.SCHEMA_KEYED_MESSAGE_KEY))
79+
|| "true".equals(context.getJMeterVariables().get(PropsKeysHelper.SIMPLE_KEYED_MESSAGE_KEY))) {
80+
keyMessageFlag = true;
81+
if (!Objects.isNull(JMeterContextService.getContext().getVariables().get(PropsKeysHelper.KEY_SUBJECT_NAME))) {
82+
keyGenerator = SamplerUtil.configureKeyGenerator(props);
8983
} else {
90-
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ProducerKeysHelper.KEY_SERIALIZER_CLASS_CONFIG_DEFAULT);
91-
}
92-
93-
if (context.getParameter(ProducerKeysHelper.APICURIO_LEGACY_ID_HANDLER).equals(ProducerKeysHelper.FLAG_YES)) {
94-
props.put(SerdeConfig.ID_HANDLER, Legacy4ByteIdHandler.class.getName());
95-
}
96-
if (context.getParameter(ProducerKeysHelper.APICURIO_ENABLE_HEADERS_ID).equals(ProducerKeysHelper.FLAG_NO)) {
97-
props.put(SerdeConfig.ENABLE_HEADERS, "false");
84+
msgKeyType = props.getProperty(PropsKeysHelper.MESSAGE_KEY_KEY_TYPE);
85+
msgKeyValue = PropsKeysHelper.MSG_KEY_VALUE.equalsIgnoreCase(props.getProperty(PropsKeysHelper.MESSAGE_KEY_KEY_VALUE))
86+
? Collections.emptyList() : Collections.singletonList(props.getProperty(PropsKeysHelper.MESSAGE_KEY_KEY_VALUE));
9887
}
88+
} else {
89+
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ProducerKeysHelper.KEY_SERIALIZER_CLASS_CONFIG_DEFAULT);
90+
}
9991

100-
topic = context.getParameter(ProducerKeysHelper.KAFKA_TOPIC_CONFIG);
92+
if (context.getParameter(ProducerKeysHelper.APICURIO_LEGACY_ID_HANDLER).equals(ProducerKeysHelper.FLAG_YES)) {
93+
props.put(SerdeConfig.ID_HANDLER, Legacy4ByteIdHandler.class.getName());
94+
}
95+
if (context.getParameter(ProducerKeysHelper.APICURIO_ENABLE_HEADERS_ID).equals(ProducerKeysHelper.FLAG_NO)) {
96+
props.put(SerdeConfig.ENABLE_HEADERS, "false");
97+
}
10198

99+
topic = context.getParameter(ProducerKeysHelper.KAFKA_TOPIC_CONFIG);
100+
try {
102101

103102
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, context.getParameter(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG));
104103
props.put(ProducerConfig.CLIENT_ID_CONFIG, context.getParameter(ProducerConfig.CLIENT_ID_CONFIG));
105104
props.putIfAbsent(ProducerConfig.ACKS_CONFIG, "all");
106-
props.putIfAbsent(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
107-
props.putIfAbsent(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, AvroKafkaSerializer.class.getName());
105+
props.putIfAbsent(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ProducerKeysHelper.KEY_SERIALIZER_CLASS_CONFIG_DEFAULT);
106+
props.putIfAbsent(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ProducerKeysHelper.VALUE_SERIALIZER_CLASS_CONFIG_DEFAULT);
108107

109-
producer = new KafkaProducer<>(props);
110-
} catch (final KafkaException ex) {
108+
producer = new KafkaProducer<>(props, (Serializer) Class.forName((String) props.get(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG)).getConstructor().newInstance(),
109+
(Serializer) Class.forName((String) props.get(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG)).getConstructor().newInstance());
110+
} catch (final KafkaException | ClassNotFoundException ex) {
111111
getNewLogger().error(ex.getMessage(), ex);
112+
} catch (InvocationTargetException | NoSuchMethodException | InstantiationException | IllegalAccessException e) {
113+
throw new KLoadGenException(e);
112114
}
113115
}
114116

@@ -135,7 +137,7 @@ public SampleResult runTest(final JavaSamplerContext javaSamplerContext) {
135137

136138
if (Objects.nonNull(messageVal)) {
137139
try {
138-
final var producerRecord = getProducerRecord(messageVal, enrichedKeyFlag(), enrichedValueFlag());
140+
final var producerRecord = getProducerRecord(messageVal);
139141
final var headersSB = new ArrayList<>(SamplerUtil.populateHeaders(kafkaHeaders, producerRecord));
140142

141143
sampleResult.setRequestHeaders(StringUtils.join(headersSB, ","));
@@ -169,30 +171,26 @@ private List<HeaderMapping> safeGetKafkaHeaders(final JMeterContext jmeterContex
169171
return headerMappingList;
170172
}
171173

172-
private ProducerRecord<Object, Object> getProducerRecord(final EnrichedRecord messageVal, final boolean keyFlag, final boolean valueFlag) {
174+
private ProducerRecord<Object, Object> getProducerRecord(final EnrichedRecord messageVal) {
173175
final ProducerRecord<Object, Object> producerRecord;
176+
177+
final String keySerializer = (String) props.get(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG);
178+
final String valueSerializer = (String) props.get(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG);
179+
174180
if (keyMessageFlag) {
175181
if (Objects.isNull(keyGenerator)) {
176-
final var key = statelessGeneratorTool.generateObject("key", msgKeyType, 0, msgKeyValue).toString();
177-
producerRecord = new ProducerRecord<>(topic, key, getObject(messageVal, valueFlag));
182+
final var key = statelessGeneratorTool.generateObject("key", msgKeyType, 0, msgKeyValue);
183+
producerRecord = new ProducerRecord<>(topic, key, getObject(messageVal, valueSerializer));
178184
} else {
179185
final var key = keyGenerator.nextMessage();
180-
producerRecord = new ProducerRecord<>(topic, getObject(key, keyFlag), getObject(messageVal, valueFlag));
186+
producerRecord = new ProducerRecord<>(topic, getObject(key, keySerializer), getObject(messageVal, valueSerializer));
181187
}
182188
} else {
183-
producerRecord = new ProducerRecord<>(topic, getObject(messageVal, valueFlag));
189+
producerRecord = new ProducerRecord<>(topic, getObject(messageVal, valueSerializer), getObject(messageVal, valueSerializer));
184190
}
185191
return producerRecord;
186192
}
187193

188-
private Boolean enrichedKeyFlag() {
189-
return SERIALIZER_SET.contains(props.get(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG).toString());
190-
}
191-
192-
private Boolean enrichedValueFlag() {
193-
return SERIALIZER_SET.contains(props.get(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG).toString());
194-
}
195-
196194
private void fillSamplerResult(final ProducerRecord<Object, Object> producerRecord, final SampleResult sampleResult) {
197195
final String result = "key: "
198196
+ producerRecord.key()
@@ -220,8 +218,8 @@ private String prettyPrint(final RecordMetadata recordMetadata) {
220218
return String.format(TEMPLATE, recordMetadata.topic(), recordMetadata.partition(), recordMetadata.offset());
221219
}
222220

223-
private Object getObject(final EnrichedRecord messageVal, final boolean isKloadSerializer) {
224-
return isKloadSerializer ? messageVal : messageVal.getGenericRecord();
221+
private Object getObject(final EnrichedRecord messageVal, final String serializer) {
222+
return (serializer.contains("com.sngular.kloadgen") && !serializer.contains("Generic")) ? messageVal : messageVal.getGenericRecord();
225223
}
226224
}
227225

‎src/main/java/com/sngular/kloadgen/sampler/SamplerUtil.java‎

Lines changed: 1 addition & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -395,13 +395,7 @@ public static BaseLoadGenerator configureValueGenerator(final Properties props)
395395
}
396396
} else {
397397
try {
398-
final String schema;
399-
if (jMeterVariables.get(PropsKeysHelper.VALUE_SCHEMA).isEmpty()) {
400-
schema = props.getProperty(PropsKeysHelper.VALUE_SCHEMA);
401-
} else {
402-
schema = jMeterVariables.get(PropsKeysHelper.VALUE_SCHEMA);
403-
}
404-
generator.setUpGenerator(schema, (List<FieldValueMapping>) jMeterVariables.getObject(PropsKeysHelper.VALUE_SCHEMA_PROPERTIES));
398+
generator.setUpGenerator(jMeterVariables.get(PropsKeysHelper.VALUE_SCHEMA), (List<FieldValueMapping>) jMeterVariables.getObject(PropsKeysHelper.VALUE_SCHEMA_PROPERTIES));
405399
} catch (final SchemaParseException exc) {
406400
generator.setUpGenerator(jMeterVariables.get(PropsKeysHelper.VALUE_SCHEMA), (List<FieldValueMapping>) jMeterVariables.getObject(PropsKeysHelper.VALUE_SCHEMA_PROPERTIES));
407401
}

‎src/main/java/com/sngular/kloadgen/util/ProducerKeysHelper.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -6,8 +6,8 @@
66

77
package com.sngular.kloadgen.util;
88

9+
import com.sngular.kloadgen.serializer.AvroSerializer;
910
import com.sngular.kloadgen.serializer.GenericAvroRecordSerializer;
10-
import org.apache.kafka.common.serialization.StringSerializer;
1111

1212
public final class ProducerKeysHelper {
1313

@@ -19,7 +19,7 @@ public final class ProducerKeysHelper {
1919

2020
public static final String KAFKA_TOPIC_CONFIG_DEFAULT = "<Topic>";
2121

22-
public static final String KEY_SERIALIZER_CLASS_CONFIG_DEFAULT = StringSerializer.class.getName();
22+
public static final String KEY_SERIALIZER_CLASS_CONFIG_DEFAULT = AvroSerializer.class.getName();
2323

2424
public static final String VALUE_SERIALIZER_CLASS_CONFIG_DEFAULT = GenericAvroRecordSerializer.class.getName();
2525

0 commit comments

Comments
 (0)