KStreamBuilder builder = new KStreamBuilder();
		Map<String, Object> serdeProps = new HashMap<>();
		Serializer<Message> messageSerializer = new JsonPOJOSerializer<>();
		serdeProps.put("JsonPOJOClass", Message.class);
		messageSerializer.configure(serdeProps, false);

		Deserializer<Message> messageDeserializer = new JsonPOJODeserializer<>();
		serdeProps.put("JsonPOJOClass", Message.class);
		messageDeserializer.configure(serdeProps, false);

		Serde<Message> messageSerde = Serdes.serdeFrom(messageSerializer, messageDeserializer);
		builder.stream(Serdes.Long(), messageSerde, "cancelSP")
				.mapValues(value -> {
					return new Message(...);
				}).to("cancelSP");