Make KStream health result serializable (#1026)

[tmellanxt]

Summary

This pull request addresses issue #1026 by implementing a custom Serde<TaskId> to kafka-streams module. This avoid a serialization error when calling the /health endpoint, which throws in turn a:

No serializable introspection present for type TaskId. Consider adding Serdeable. Serializable annotate to type TaskId. Alternatively if you are not in control of the project's source code, you can use @SerdeImport(TaskId.class) to enable serialization of this type.

Changes Made

• Added serde processor as annotationProcessor, and serde api as compileOnly to load in the annotation.
• Created a class dedicated to adding annotation for external classes that need serialising as part of the health result

Related Issues

• Fixes Kafka Streams and Jackson Serialization results in CodecException when calling Health endpoint #1026

[graemerocher]

thanks can you make this non-public and add @Internal?

[tmellanxt]

Good point, done

[graemerocher]

thanks, a test would be good if possible

[tmellanxt]

I had written one, but wasn't sure if it was too specific a test to use a JsonMapper to check if a TaskId is serializable. If you think its appropriate here I'll add it.

The kafka-streams instances injected into the tests have 0 active or standby tasks which is why this was never caught.

Lmk your thoughts on this. If you want the concrete test I'll add it

[graemerocher]

just a check that the type is serialisable is fine

[tmellanxt]

While testing I've realised that the TaskId isn't serializable out of the box with Jackson, and if it were it would yield:

"taskId": {
  "topicGroupId": 0,
  "partition": 0
}

but I'm not sure this is helpful as part of the health report however.

The TaskId object has a custom .toString() which formats it in a readable manner. I think its more appropriate here.

"taskId": "0_0"

After all, the task id is meant to be an identifier, showing its component internal fields in this manner doesn't seem helpful.

This would be the change, adding the toString(). I'd make it null safe of course

details.put("taskId", taskMetadata.taskId().toString());

Thoughts?

[graemerocher]

sure, for that case you probably need to implement a Serializer and remove the @SerdeImport

[tmellanxt]

Well I was just going to call .toString() on the task object that gets put into the Map<String, Object>.

Or would you rather a custom Serializer to allow users to override the serializer if they want?

[tmellanxt]

Went with the custom Serde

[graemerocher]

sorry would you mind rebasing and then I will merge

[tmellanxt]

sorry would you mind rebasing and then I will merge

Rebased @graemerocher

[graemerocher]

Thanks for the contribution!