Apply typed schema to cached messages; `kafka cache fetch server`

[jfallows]

Verify received message conforms to expected type, handle non-conforming messages via ignore, log or dead letter queue.