public void send(String key, String value) {
        ProducerRecord<String, String> record;

        // Older versions of ProducerRecord don't include the message create time in the constructor. So including
        // even a 'null' argument results in a NoSuchMethodException. Thus we only include the create time if it is
        // explicitly specified to remain fully backward compatible with older clients.
        if (createTime != null) {
            record = new ProducerRecord<>(topic, null, createTime, key, value);
            createTime += System.currentTimeMillis() - startTime;
        } else {
            record = new ProducerRecord<>(topic, key, value);
        }

        numSent++;
        try {
            producer.send(record, new PrintInfoCallback(key, value));
        } catch (Exception e) {

            synchronized (System.out) {
                printJson(new FailedSend(key, value, topic, e));
            }
        }
    }
Produce a message with given key and value.
tools/src/main/java/org/apache/kafka/tools/VerifiableProducer.java