If you pass in the classOf the log and continue exception handler, the mocked streams fails because it can't cast this to a string. KafkaStreams takes care of this internally.
p.put(DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
classOf[LogAndContinueExceptionHandler])
KafkaStream's Properties is a Hashtable<Object, Object>
This is with version 3.4.0