authorjapps / authorjapps/zerocode
Allow dynamic creation of records for kafka produce
- 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
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