private RecordMetadata syncSend(KafkaProducer<Integer, String> producer, int key, String value)
        throws ExecutionException, InterruptedException {
        try {
            // send the record and then call get, which blocks waiting for the ack from the broker
            RecordMetadata metadata = producer.send(new ProducerRecord<>(topic, key, value)).get();
            Utils.maybePrintRecord(numRecords, key, value, metadata);
            return metadata;
        } catch (AuthorizationException | UnsupportedVersionException | ProducerFencedException
                 | FencedInstanceIdException | OutOfOrderSequenceException | SerializationException e) {
            Utils.printErr(e.getMessage());
            // we can't recover from these exceptions
            shutdown();
        } catch (KafkaException e) {
            Utils.printErr(e.getMessage());
        }
        return null;
    }
A simple producer thread supporting two send modes:
- Async mode (default): records are sent without waiting for the response.
- Sync mode: each send operation blocks waiting for the response.
examples/src/main/java/kafka/examples/Producer.java