ProducerRecord should validate the topic name
Author: huxihx
Reviewers: Ismael Juma
Closes from huxihx/KAFKA-5098_Does_not_check_topic_name_before_sending
*/
package org.apache.kafka.clients.producer;
+import org.apache.kafka.common.errors.InvalidTopicException;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.header.internals.RecordHeaders;
+import org.apache.kafka.common.internals.Topic;
/**
* A key/value pair to be sent to Kafka. This consists of a topic name to which the record is being sent, an optional
@@ -62,10 +64,17 @@ public class ProducerRecord<K, V> {
* @param key The key that will be included in the record
* @param value The record contents
* @param headers the headers that will be included in the record
+ * @throws IllegalArgumentException if the topic name is null or invalid
+ * @throws IllegalArgumentException if the partition or timestamp is negative
*/
public ProducerRecord(String topic, Integer partition, Long timestamp, K key, V value, Iterable<Header> headers) {
if (topic == null)
throw new IllegalArgumentException("Topic cannot be null.");
+ try {
+ Topic.validate(topic);
+ } catch (InvalidTopicException e) {
+ throw new IllegalArgumentException(e.getMessage());
+ }
if (timestamp != null && timestamp < 0)
throw new IllegalArgumentException(
String.format("Invalid timestamp: %d. Timestamp should always be non-negative or null.", timestamp));
