@Override
    public synchronized Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {
        if (this.closed) {
            throw new IllegalStateException("MockProducer is already closed.");
        }

        if (this.producerFenced) {
            throw new KafkaException("MockProducer is fenced.", new ProducerFencedException("Fenced"));
        }
        if (this.sendException != null) {
            throw this.sendException;
        }

        int partition = 0;
        if (!this.cluster.partitionsForTopic(record.topic()).isEmpty())
            partition = partition(record, this.cluster);
        else {
            //just to throw ClassCastException if serializers are not the proper ones to serialize key/value
            keySerializer.serialize(record.topic(), record.key());
            valueSerializer.serialize(record.topic(), record.value());
        }
            
        TopicPartition topicPartition = new TopicPartition(record.topic(), partition);
        ProduceRequestResult result = new ProduceRequestResult(topicPartition);
        FutureRecordMetadata future = new FutureRecordMetadata(result, 0, RecordBatch.NO_TIMESTAMP,
                0, 0, Time.SYSTEM);
        long offset = nextOffset(topicPartition);
        long baseOffset = Math.max(0, offset - Integer.MAX_VALUE);
        int batchIndex = (int) Math.min(Integer.MAX_VALUE, offset);
        Completion completion = new Completion(offset, new RecordMetadata(topicPartition, baseOffset, batchIndex,
                RecordBatch.NO_TIMESTAMP, 0, 0), result, callback, topicPartition);

        if (!this.transactionInFlight)
            this.sent.add(record);
        else
            this.uncommittedSends.add(record);

        if (autoComplete)
            completion.complete(null);
        else
            this.completions.addLast(completion);

        return future;
    }
Adds the record to the list of sent records.

@see #history()
clients/src/main/java/org/apache/kafka/clients/producer/MockProducer.java