@Override
    public void run() {
        int key = 0;
        int sentRecords = 0;
        // the producer instance is thread safe
        try (KafkaProducer<Integer, String> producer = createKafkaProducer()) {
            while (!closed && sentRecords < numRecords) {
                if (isAsync) {
                    asyncSend(producer, key, "test" + key);
                } else {
                    syncSend(producer, key, "test" + key);
                }
                key++;
                sentRecords++;
            }
        } catch (Throwable e) {
            Utils.printErr("Unhandled exception");
            e.printStackTrace();
        }
        Utils.printOut("Sent %d records", sentRecords);
        shutdown();
    }
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