import kafka.common.FailedToSendMessageException; //导入依赖的package包/类
/**
 * Send a set of messages. Each must have a "key" string value.
 * 
 * @param topic
 * @param msg
 * @throws FailedToSendMessageException
 * @throws JSONException
 */
@Override
public void sendMessages(String topic, List<? extends message> msgs)
		throws IOException, FailedToSendMessageException {
	log.info("sending " + msgs.size() + " events to [" + topic + "]");

	final List<KeyedMessage<String, String>> kms = new ArrayList<KeyedMessage<String, String>>(msgs.size());
	for (message o : msgs) {
		final KeyedMessage<String, String> data = new KeyedMessage<String, String>(topic, o.getKey(), o.toString());
		kms.add(data);
	}
	try {
		fProducer.send(kms);

	} catch (FailedToSendMessageException excp) {
		log.error("Failed to send message(s) to topic [" + topic + "].", excp);
		throw new FailedToSendMessageException(excp.getMessage(), excp);
	}
}
 

import kafka.common.FailedToSendMessageException; //导入依赖的package包/类
protected void
output(Collection<TaskData.ExecutionPlan.Node.Port> outputPorts,
       TaskData.ExecutionPlan executionPlan,
       Serializable result,
       UUID taskID) throws Exception {
    new RobustExecutor<Void, Void>(
            () -> {
                if (verbose) {
                    KafkaHelper.sendWithLog(taskID.toString(),
                            new TaskData(outputPorts, executionPlan, result),
                            producerSingleton.getInst(),
                            loggerSingleton.getInst());
                } else {
                    KafkaHelper.send(taskID.toString(),
                            new TaskData(outputPorts, executionPlan, result),
                            producerSingleton.getInst());
                }
            },
            Arrays.asList(
                    MessageSizeTooLargeException.class,
                    KafkaException.class,
                    FailedToSendMessageException.class)
    ).execute();
}
 

import kafka.common.FailedToSendMessageException; //导入依赖的package包/类
/**
 * method publishing batch messages
 * 
 * @param topic
 * @param kms
 * throws IOException
 */
public void sendBatchMessage(String topic, ArrayList<KeyedMessage<String, String>> kms) throws IOException {
	try {
		fProducer.send(kms);

	} catch (FailedToSendMessageException excp) { 
		log.error("Failed to send message(s) to topic [" + topic + "].", excp);
		throw new FailedToSendMessageException(excp.getMessage(), excp);
	}

}
 

import kafka.common.FailedToSendMessageException; //导入依赖的package包/类
@Override
public void handle(Seq<KeyedMessage<String, String>> messages) {
    try {
        super.handle(messages);
    } catch (FailedToSendMessageException e) {
        KruxStdLib.STATSD.count("dropped_messages", messages.size());
    }

}
 

import kafka.common.FailedToSendMessageException; //导入依赖的package包/类
private boolean sendOrReportFailure(String topic, final byte[] msg) {
    final boolean nonNullTopic = !("null".equals(topic));
    if (nonNullTopic) {
        try {
            producer.send(new KeyedMessage<byte[], byte[]>(topic, msg));
        } catch (FailedToSendMessageException e) {
            log.warn("Failed to send to topic {}", topic, e);
            failedToSendMessageExceptions.incrementAndGet();
        }
    }
    return nonNullTopic;
}
 

import kafka.common.FailedToSendMessageException; //导入依赖的package包/类
/**
 * Sends messages to Kafka topic using specified properties in kafka.properties file.
 *
 * @param producer kafka producer provided by the caller
 * @param event    event that should be sent to Kafka Broker
 */
protected void sendMessageToKafka(Producer producer, Message<JsonObject> event) {

    if (!isValid(event.body().getString(PAYLOAD))) {
        logger.error("Invalid kafka message provided. Message not sent to kafka...");
        sendError(event, String.format("Invalid kafka message provided. Property [%s] is not set.", PAYLOAD)) ;
        return;
    }

    try {
        final MessageHandler messageHandler = messageHandlerFactory.createMessageHandler(serializerType);

        String topic = isValid(event.body().getString(TOPIC)) ? event.body().getString(TOPIC) : getTopic();     
        String partKey = event.body().getString(PART_KEY);

        long startTime = System.currentTimeMillis();
        
        messageHandler.send(producer, topic, partKey, event.body());
        
        statsDClient.recordExecutionTime("submitted", (System.currentTimeMillis()-startTime));
        
        
        sendOK(event);
        logger.info(String.format("Message sent to kafka topic: %s. Payload: %s", topic, event.body().getString(PAYLOAD)));
    } catch (FailedToSendMessageException ex) {
        sendError(event, "Failed to send message to Kafka broker...", ex);
    }
}
 

import kafka.common.FailedToSendMessageException; //导入依赖的package包/类
@Test(expected = FailedToSendMessageException.class)
public void sendMessageStatsDDisabled() throws Exception {
    JsonObject jsonObject = new JsonObject();
    jsonObject.putString(EventProperties.PAYLOAD, MESSAGE);

    Handler<Message<JsonObject>> replyHandler = new Handler<Message<JsonObject>>() {
        public void handle(Message<JsonObject> message) {
            assertEquals("error", message.body().getString("status"));
            assertTrue(message.body().getString("message").equals("Failed to send message to Kafka broker..."));
            testComplete();
        }
    };
    vertx.eventBus().send(ADDRESS, jsonObject, replyHandler);
}
 

import kafka.common.FailedToSendMessageException; //导入依赖的package包/类
@Test(expected = FailedToSendMessageException.class)
public void sendMessage() throws Exception {
    JsonObject jsonObject = new JsonObject();
    jsonObject.putBinary(EventProperties.PAYLOAD, MESSAGE.getBytes());

    Handler<Message<JsonObject>> replyHandler = new Handler<Message<JsonObject>>() {
        public void handle(Message<JsonObject> message) {
            assertEquals("error", message.body().getString("status"));
            assertTrue(message.body().getString("message").equals("Failed to send message to Kafka broker..."));
            testComplete();
        }
    };
    vertx.eventBus().send(ADDRESS, jsonObject, replyHandler);
}
 

import kafka.common.FailedToSendMessageException; //导入依赖的package包/类
@Test(expected = FailedToSendMessageException.class)
public void sendMessage() throws Exception {
    JsonObject jsonObject = new JsonObject();
    jsonObject.putString(EventProperties.PAYLOAD, MESSAGE);

    Handler<Message<JsonObject>> replyHandler = new Handler<Message<JsonObject>>() {
        public void handle(Message<JsonObject> message) {
            assertEquals("error", message.body().getString("status"));
            assertTrue(message.body().getString("message").equals("Failed to send message to Kafka broker..."));
            testComplete();
        }
    };
    vertx.eventBus().send(ADDRESS, jsonObject, replyHandler);
}
 

import kafka.common.FailedToSendMessageException; //导入依赖的package包/类
/**
 * Send a message with a given topic and key.
 * 
 * @param msg
 * @throws FailedToSendMessageException
 * @throws JSONException
 */
@Override
public void sendMessage(String topic, message msg) throws IOException, FailedToSendMessageException {
	final List<message> msgs = new LinkedList<message>();
	msgs.add(msg);
	sendMessages(topic, msgs);
}