Add string and byte serializer

[thmshmm]

fixes #27

Added support to serialize string and byte array keys/values. Also fixes producing Protobuf data when the data is converted to a byte array upfront.

[mostafa]

Thanks for your contribution. 👏 🙏
Overall LGTM, but needs a small change. I have a question: might a user need to receive a byte array while consuming messages?

[thmshmm]

Yes, deserialization should be implemented as well. I can work on it if I have some time. For now the regular consumer can be used to "just" read some messages and check if they were written.

[mostafa]

@thmshmm Let's wait until you add that part, too, before merging the PR.

[thmshmm]

@mostafa can you review again pls? Added the deserialisation part as well.

[mostafa]

Hey @thmshmm,

Thanks again for your contribution. Have you tested this code? I tried it, and it fails on a check:

✗ payload is correct
↳  0% — ✓ 0 / ✗ 7

And the reason is that the returned data is a string, not a byte array.

[thmshmm]

It works for me ✓ payload is correct. I also noticed this, but found no solution. Thats why I used String.fromCharCode(...msgs[0].value.split(",")) to convert it back. I think this behaviour may come from the goja implementation that k6 uses.

I use it to serialize Protobuf (OpenTelemetry) messages which also works.

[mostafa]

I used your script called scripts/test_bytes.js, which doesn't work. I will try again to see if I can find a solution for it.

[thmshmm]

Do you have any errors when creating the k6 bin?
xk6 build --with github.com/thmshmm/xk6-kafka@1fd4880ca75cdf1c78930e419d192e3bf36397b82 --replace github.com/thmshmm/xk6-kafka@latest=github.com/mostafa/xk6-kafka@latest --output k6

[mostafa]

@thmshmm I don't have any errors, and I tested it with k6 v0.36.0 and v0.37.0 (by updating go.mod and running go mod tidy -go=1.18). This is the result of running ./k6 run scripts/test_bytes.js:

$ ./k6 run scripts/test_bytes.js

          /\      |‾‾| /‾‾/   /‾‾/
     /\  /  \     |  |/  /   /  /
    /  \/    \    |     (   /   ‾‾\
   /          \   |  |\  \ |  (‾)  |
  / __________ \  |__| \__\ \_____/ .io

WARN[0000] Module 'k6/x/kafka' is using deprecated APIs that will be removed in k6 v0.38.0, for more details on how to update it see https://k6.io/docs/extensions/guides/create-an-extension/#advanced-javascript-extension
  execution: local
     script: scripts/test_bytes.js
     output: -

  scenarios: (100.00%) 1 scenario, 1 max VUs, 10m30s max duration (incl. graceful stop):
           * default: 1 iterations for each of 1 VUs (maxDuration: 10m0s, gracefulStop: 30s)


running (00m00.1s), 0/1 VUs, 1 complete and 0 interrupted iterations
default ✓ [======================================] 1 VUs  00m00.1s/10m0s  1/1 iters, 1 per VU

     ✓ is sent
     ✓ 10 messages returned
     ✓ key starts with 'test-id-' string
     ✗ payload is correct
      ↳  0% — ✓ 0 / ✗ 1

     █ teardown

     checks.........................: 99.02% ✓ 102        ✗ 1
     data_received..................: 0 B    0 B/s
     data_sent......................: 0 B    0 B/s
     iteration_duration.............: avg=44.78ms min=27.23µs med=44.78ms max=89.54ms p(90)=80.58ms p(95)=85.06ms
     iterations.....................: 1      10.656402/s
     kafka.reader.dial.count........: 1      10.656402/s
     kafka.reader.error.count.......: 0      0/s
     kafka.reader.fetches.count.....: 1      10.656402/s
     kafka.reader.message.bytes.....: 372 B  4.0 kB/s
     kafka.reader.message.count.....: 12     127.876822/s
     kafka.reader.rebalance.count...: 0      0/s
     kafka.reader.timeouts.count....: 0      0/s
     kafka.writer.dial.count........: 2      21.312804/s
     kafka.writer.error.count.......: 0      0/s
     kafka.writer.message.bytes.....: 11 kB  115 kB/s
     kafka.writer.message.count.....: 200    2131.28037/s
     kafka.writer.rebalance.count...: 0      0/s
     kafka.writer.write.count.......: 200    2131.28037/s

[mostafa]

@thmshmm Just found the problem! I'll make suggestions, so you can merge them right away into your PR.

[mostafa]

@thmshmm
These are the changes you need to make to make the script work as expected. As far as I see, the implementation works as expected, but the script should be fixed.

[thmshmm]

@mostafa thanks for your support, I integrated all your corrections. But, for me line 62 only works like this
"payload is correct": (msgs) => String.fromCharCode(...msgs[0].value.split(",")) === payload, with the .split(","). Do you have any idea?

It is definitely the byte deserializer that is called for the value, but when I check typeof(message[0].value) the result is string

[mostafa]

@thmshmm Thanks again for your contribution! I'll merge this and release a version after that.

As for your question, I tried these after let messages = consumeWithConfiguration(consumer, 10, configuration);:

console.log(messages[0].value);
console.log(typeof messages[0].value);
console.log(typeof messages[0].value[0]);

and this is the k6 console output:

INFO[0000] 98,121,116,101,32,97,114,114,97,121,32,112,97,121,108,111,97,100  source=console
INFO[0000] object                                        source=console
INFO[0000] number                                        source=console

I also tried String.fromCharCode, and it only accepts a sequence (array?) of numbers and returns a String. Anything else isn't supported and will result in an empty string.