@Override
    public <K, V> void send(final String topic,
                            final K key,
                            final V value,
                            final Headers headers,
                            final Integer partition,
                            final Long timestamp,
                            final Serializer<K> keySerializer,
                            final Serializer<V> valueSerializer,
                            final String processorNodeId,
                            final InternalProcessorContext<Void, Void> context) {
        checkForException();

        final byte[] keyBytes;
        final byte[] valBytes;
        try {
            keyBytes = keySerializer.serialize(topic, headers, key);
            valBytes = valueSerializer.serialize(topic, headers, value);
        } catch (final ClassCastException exception) {
            final String keyClass = key == null ? "unknown because key is null" : key.getClass().getName();
            final String valueClass = value == null ? "unknown because value is null" : value.getClass().getName();
            throw new StreamsException(
                String.format(
                    "ClassCastException while producing data to topic %s. " +
                        "A serializer (key: %s / value: %s) is not compatible to the actual key or value type " +
                        "(key type: %s / value type: %s). " +
                        "Change the default Serdes in StreamConfig or provide correct Serdes via method parameters " +
                        "(for example if using the DSL, `#to(String topic, Produced<K, V> produced)` with " +
                        "`Produced.keySerde(WindowedSerdes.timeWindowedSerdeFrom(String.class))`).",
                    topic,
                    keySerializer.getClass().getName(),
                    valueSerializer.getClass().getName(),
                    keyClass,
                    valueClass),
                exception);
        } catch (final Exception exception) {
            final ProducerRecord<K, V> record = new ProducerRecord<>(topic, partition, timestamp, key, value, headers);
            final ProductionExceptionHandler.ProductionExceptionHandlerResponse response;

            log.debug(String.format("Error serializing record to topic %s", topic), exception);

            try {
                response = productionExceptionHandler.handleSerializationException(record, exception);
            } catch (final Exception e) {
                log.error("Fatal when handling serialization exception", e);
                recordSendError(topic, e, null);
                return;
            }

            if (response == ProductionExceptionHandlerResponse.FAIL) {
                throw new StreamsException(
                    String.format(
                        "Unable to serialize record. ProducerRecord(topic=[%s], partition=[%d], timestamp=[%d]",
                        topic,
                        partition,
                        timestamp),
                    exception
                );
            }

            log.warn("Unable to serialize record, continue processing. " +
                            "ProducerRecord(topic=[{}], partition=[{}], timestamp=[{}])",
                    topic,
                    partition,
                    timestamp);

            droppedRecordsSensor.record();

            return;
        }

        final ProducerRecord<byte[], byte[]> serializedRecord = new ProducerRecord<>(topic, partition, timestamp, keyBytes, valBytes, headers);

        streamsProducer.send(serializedRecord, (metadata, exception) -> {
            // if there's already an exception record, skip logging offsets or new exceptions
            if (sendException.get() != null) {
                return;
            }

            if (exception == null) {
                final TopicPartition tp = new TopicPartition(metadata.topic(), metadata.partition());
                if (metadata.offset() >= 0L) {
                    offsets.put(tp, metadata.offset());
                } else {
                    log.warn("Received offset={} in produce response for {}", metadata.offset(), tp);
                }

                if (!topic.endsWith("-changelog")) {
                    // we may not have created a sensor during initialization if the node uses dynamic topic routing,
                    // as all topics are not known up front, so create the sensor for this topic if absent
                    final Sensor topicProducedSensor = producedSensorByTopic.computeIfAbsent(
                        topic,
                        t -> TopicMetrics.producedSensor(
                            Thread.currentThread().getName(),
                            taskId.toString(),
                            processorNodeId,
                            topic,
                            context.metrics()
                        )
                    );
                    final long bytesProduced = producerRecordSizeInBytes(serializedRecord);
                    topicProducedSensor.record(bytesProduced, context.currentSystemTimeMs());
                }
            } else {
                recordSendError(topic, exception, serializedRecord);

                // KAFKA-7510 only put message key and value in TRACE level log so we don't leak data by default
                log.trace("Failed record: (key {} value {} timestamp {}) topic=[{}] partition=[{}]", key, value, timestamp, topic, partition);
            }
        });
    }
@throws StreamsException fatal error that should cause the thread to die
@throws TaskMigratedException recoverable error that would cause the task to be removed
streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java