authorjapps / authorjapps/zerocode

Allow dynamic creation of records for kafka produce

Open
#483 0 comments 0 reactions 0 assignees View on GitHub
kafka
Dominant language
Java
Stars
1k
Forks
453
Avg merge
7d 2h
Merged PRs (30d)
5

Description

Not sure if this is a feature request or a bug.

Business use-case example:
In an application a kafka messages contains references to a third party e.g. object store.
When I store multiple objects in scenario step 1, I want not to send multiple kafka records in scenario step 2 via KafkaSender.

We tried this with two different approaches:
1) generating a file with the events in step 1 and using this file as records file in step 2
2) passing the records between steps as response/request objects

**Generating a file and using that referenced file**
We already use that for some custom steps, but it didn't work for us with the `KafkaSender.java` as:
```
private File validateAndGetFile(String fileName) {
try {
URL resource = getClass().getClassLoader().getResource(fileName);
return new File(resource.getFile());
} catch (Exception ex) {
throw new RuntimeException("Error accessing file: `" + fileName + "' - " + ex);
}
}
```
But allows us only to access files which are already bundled in the resources for that test when the test starts. In other words, only the second test call succeeds, as the first is unable to find the file.

I think a solution could be to not retrieve the URL via the class loader.getResources.

**Passing the records between steps as response/request objects**
That was some very good input we got from @sparrowV via slack. But ran into serialisation issues between the steps.

The custom producing Kotlin code:
```kotlin
data class Record(val key: String, val value: JsonValue)
data class JsonValue(val name: String, val number: Number)
class SomeClassName {
fun create_my_testdata(args: Map): Map {
return mutableMapOf(
"success" to true,
"body" to listOf(
Record("1", JsonValue("First", 1)),
Record("2", JsonValue("Second", 2)),
)
)
}
}
```
the referencing kafka produce step:
```
{
"name": "load_kafka",
"url": "kafka-topic:kafkaTopic",
"operation": "produce",
"request": {
"async": false,
"recordType": "JSON",
"records": "${$.create_testdata.response.body}"
},
"assertions": {
"status": "Ok",
"recordMetadata": "$NOT.NULL"
}
}
```

Resulted in following log:
```
2021-02-23 16:04:41,459 [main] ERROR org.jsmart.zerocode.core.kafka.send.KafkaSender - Error in sending record.
com.fasterxml.jackson.databind.exc.MismatchedInputException: Cannot deserialize instance of `java.util.ArrayList` out of VALUE_STRING token
at [Source: (String)"{"async":false,"recordType":"JSON","records":"[{\"key\":\"1\",\"value\":{\"name\":\"First\",\"number\":1}},{\"key\":\"2\",\"value\":{\"name\":\"Second\",\"number\":2}}]"}"; line: 1, column: 46] (through reference chain: org.jsmart.zerocode.core.kafka.send.message.ProducerJsonRecords["records"])
```
at:
```java
case JSON:
jsonRecords = objectMapper.readValue(requestJson, ProducerJsonRecords.class);

fileName = jsonRecords.getFile();
if (fileName != null) {
File file = validateAndGetFile(fileName);
try (BufferedReader br = new BufferedReader(new FileReader
```

In the `KafkaSender.java` `send()` method, at this line ` jsonRecords = objectMapper.readValue(requestJson, ProducerJsonRecords.class);`.

When I changed the requestJson during debugging to not quote the array value for records it worked fine.
So my current understanding is that in `ZeroCodeAssertionsProcessorImpl.java`:
```java
@Override
public String resolveJsonPaths(String jsonString, String scenarioState) {

...

} else {
Object jsonPathValue = JsonPath.read(scenarioState, thisPath);
if (isPathValueJson(jsonPathValue)) {
final String jsonAsString = mapper.writeValueAsString(jsonPathValue);
String escapedJsonString = escapeJava(jsonAsString);
paramMap.put(thisPath, escapedJsonString);

} else {

paramMap.put(thisPath, JsonPath.read(scenarioState, thisPath));

}
}
```
the parameter `jsonString` already contains quotation for the variable:
`{"async":false,"recordType":"JSON","records":"${$.create_testdata.response.body}"}`
So when `${$.create_testdata.response.body}` will be substituted the surrounding quotation stays. Which breaks the deserialisation in the following step.

I think a solution could be to add a special case like rawBody, which also removes the quotation and so allowing to store complex json structures.

For now we settle with executing the test twice, ignoring that it fails on the first run. But I would really like an upstream way to this :).

Contributor guide

Open the contributing guide

Research direction

Start with KafkaSender.java, especially send() and the ProducerJsonRecords deserialization path, then inspect ZeroCodeAssertionsProcessorImpl.java and resolveJsonPaths(). Reproduce the quoted records substitution described in the issue and trace both the generated-file and response/request approaches. Done means dynamically produced records can be passed to KafkaSender and deserialized successfully without breaking the existing file-based behavior.

Written by the indexing model from the issue text.

Assessment

Tech stack
java, kafka
Domain
stream-processing, testing
Issue type
Feature
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.