@Override
    public void init(final InternalProcessorContext<Void, Void> context) {
        super.init(context);
        this.context = context;
        try {
            keySerializer = prepareKeySerializer(keySerializer, context, this.name());
        } catch (ConfigException | StreamsException e) {
            throw new StreamsException(String.format("Failed to initialize key serdes for sink node %s", name()), e, context.taskId());
        }

        try {
            valSerializer = prepareValueSerializer(valSerializer, context, this.name());
        } catch (final ConfigException | StreamsException e) {
            throw new StreamsException(String.format("Failed to initialize value serdes for sink node %s", name()), e, context.taskId());
        }
    }
@throws UnsupportedOperationException if this method adds a child to a sink node
streams/src/main/java/org/apache/kafka/streams/processor/internals/SinkNode.java