Using Producer API to Produce Messages for Security Topics

Function Description

It is used by the Producer APIs to produce messages for the security topic.

Code Sample

The following is a code snippet used by the Producer API to produce messages to a secure topic.

The following code snippet belongs to the run method of the com.huawei.bigdata.kafka.example.Producer class.

/** 
 * The Producer thread executes a function to send messages periodically. 
 */ 
public void run() {
    LOG.info("New Producer: start.");
    int messageNo = 1;
    
    while (messageNo <= MESSAGE_NUM) {
        String messageStr = "Message_" + messageNo;
        long startTime = System.currentTimeMillis();
        
        // Construct message records. 
        ProducerRecord<Integer, String> record = new ProducerRecord<Integer, String>(topic, messageNo, messageStr);
        
        if (isAsync) {
            // Sending in asynchronous mode
            producer.send(record, new DemoCallBack(startTime, messageNo, messageStr));
        } else {
            try {
                // Sending in synchronous mode
                producer.send(record).get();
                long elapsedTime = System.currentTimeMillis() - startTime;
                LOG.info("message(" + messageNo + ", " + messageStr + ") sent to topic(" + topic + ") in " + elapsedTime + " ms.");
            } catch (InterruptedException ie) {
                LOG.info("The InterruptedException occured : {}.", ie);
            } catch (ExecutionException ee) {
                LOG.info("The ExecutionException occured : {}.", ee);
            }
        }
        messageNo++;
    }
}