diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java index 4f5aa3e4726e4..15bf55f48d86f 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java +++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java @@ -25,6 +25,7 @@ import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Set; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.atomic.AtomicInteger; @@ -54,7 +55,6 @@ import org.apache.kafka.common.utils.CopyOnWriteMap; import org.apache.kafka.common.utils.LogContext; import org.apache.kafka.common.utils.Time; -import org.apache.kafka.common.utils.Utils; import org.slf4j.Logger; /** @@ -229,7 +229,8 @@ public RecordAppendResult append(TopicPartition tp, MemoryRecordsBuilder recordsBuilder = recordsBuilder(buffer, maxUsableMagic); ProducerBatch batch = new ProducerBatch(tp, recordsBuilder, time.milliseconds()); - FutureRecordMetadata future = Utils.notNull(batch.tryAppend(timestamp, key, value, headers, callback, time.milliseconds())); + FutureRecordMetadata future = Objects.requireNonNull(batch.tryAppend(timestamp, key, value, headers, + callback, time.milliseconds())); dq.addLast(batch); incomplete.add(batch); diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java b/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java index dfa6424afe863..4cd743d3d77ce 100644 --- a/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java +++ b/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java @@ -54,7 +54,6 @@ import org.apache.kafka.common.requests.RequestHeader; import org.apache.kafka.common.utils.LogContext; import org.apache.kafka.common.utils.Time; -import org.apache.kafka.common.utils.Utils; import org.slf4j.Logger; import java.io.IOException; @@ -64,6 +63,7 @@ import java.util.Iterator; import java.util.List; import java.util.Map; +import java.util.Objects; import static org.apache.kafka.common.record.RecordBatch.NO_TIMESTAMP; @@ -923,17 +923,17 @@ public void updateProduceRequestMetrics(Map> batche // per-topic record send rate String topicRecordsCountName = "topic." + topic + ".records-per-batch"; - Sensor topicRecordCount = Utils.notNull(this.metrics.getSensor(topicRecordsCountName)); + Sensor topicRecordCount = Objects.requireNonNull(this.metrics.getSensor(topicRecordsCountName)); topicRecordCount.record(batch.recordCount); // per-topic bytes send rate String topicByteRateName = "topic." + topic + ".bytes"; - Sensor topicByteRate = Utils.notNull(this.metrics.getSensor(topicByteRateName)); + Sensor topicByteRate = Objects.requireNonNull(this.metrics.getSensor(topicByteRateName)); topicByteRate.record(batch.estimatedSizeInBytes()); // per-topic compression rate String topicCompressionRateName = "topic." + topic + ".compression-rate"; - Sensor topicCompressionRate = Utils.notNull(this.metrics.getSensor(topicCompressionRateName)); + Sensor topicCompressionRate = Objects.requireNonNull(this.metrics.getSensor(topicCompressionRateName)); topicCompressionRate.record(batch.compressionRatio()); // global metrics diff --git a/clients/src/main/java/org/apache/kafka/common/MetricName.java b/clients/src/main/java/org/apache/kafka/common/MetricName.java index 0af77a038b4b4..9d695b5cf5efd 100644 --- a/clients/src/main/java/org/apache/kafka/common/MetricName.java +++ b/clients/src/main/java/org/apache/kafka/common/MetricName.java @@ -17,8 +17,7 @@ package org.apache.kafka.common; import java.util.Map; - -import org.apache.kafka.common.utils.Utils; +import java.util.Objects; /** * The MetricName class encapsulates a metric's name, logical group and its related attributes. It should be constructed using metrics.MetricName(...). @@ -78,10 +77,10 @@ public final class MetricName { * @param tags additional key/value attributes of the metric */ public MetricName(String name, String group, String description, Map tags) { - this.name = Utils.notNull(name); - this.group = Utils.notNull(group); - this.description = Utils.notNull(description); - this.tags = Utils.notNull(tags); + this.name = Objects.requireNonNull(name); + this.group = Objects.requireNonNull(group); + this.description = Objects.requireNonNull(description); + this.tags = Objects.requireNonNull(tags); } public String name() { diff --git a/clients/src/main/java/org/apache/kafka/common/MetricNameTemplate.java b/clients/src/main/java/org/apache/kafka/common/MetricNameTemplate.java index abe121f7ac972..f9c4ef5f730e6 100644 --- a/clients/src/main/java/org/apache/kafka/common/MetricNameTemplate.java +++ b/clients/src/main/java/org/apache/kafka/common/MetricNameTemplate.java @@ -21,8 +21,6 @@ import java.util.Objects; import java.util.Set; -import org.apache.kafka.common.utils.Utils; - /** * A template for a MetricName. It contains a name, group, and description, as * well as all the tags that will be used to create the mBean name. Tag values @@ -46,10 +44,10 @@ public class MetricNameTemplate { * @param tagsNames the set of metric tag names, which can/should be a set that maintains order; may not be null */ public MetricNameTemplate(String name, String group, String description, Set tagsNames) { - this.name = Utils.notNull(name); - this.group = Utils.notNull(group); - this.description = Utils.notNull(description); - this.tags = new LinkedHashSet<>(Utils.notNull(tagsNames)); + this.name = Objects.requireNonNull(name); + this.group = Objects.requireNonNull(group); + this.description = Objects.requireNonNull(description); + this.tags = new LinkedHashSet<>(Objects.requireNonNull(tagsNames)); } /** diff --git a/clients/src/main/java/org/apache/kafka/common/TopicPartitionReplica.java b/clients/src/main/java/org/apache/kafka/common/TopicPartitionReplica.java index 244260295a108..0a7c419c933bd 100644 --- a/clients/src/main/java/org/apache/kafka/common/TopicPartitionReplica.java +++ b/clients/src/main/java/org/apache/kafka/common/TopicPartitionReplica.java @@ -16,9 +16,8 @@ */ package org.apache.kafka.common; -import org.apache.kafka.common.utils.Utils; - import java.io.Serializable; +import java.util.Objects; /** @@ -32,7 +31,7 @@ public final class TopicPartitionReplica implements Serializable { private final String topic; public TopicPartitionReplica(String topic, int partition, int brokerId) { - this.topic = Utils.notNull(topic); + this.topic = Objects.requireNonNull(topic); this.partition = partition; this.brokerId = brokerId; } diff --git a/clients/src/main/java/org/apache/kafka/common/metrics/Metrics.java b/clients/src/main/java/org/apache/kafka/common/metrics/Metrics.java index 91c245dd5a186..5b331e4d42c36 100644 --- a/clients/src/main/java/org/apache/kafka/common/metrics/Metrics.java +++ b/clients/src/main/java/org/apache/kafka/common/metrics/Metrics.java @@ -20,7 +20,6 @@ import org.apache.kafka.common.MetricNameTemplate; import org.apache.kafka.common.utils.KafkaThread; import org.apache.kafka.common.utils.Time; -import org.apache.kafka.common.utils.Utils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -33,6 +32,7 @@ import java.util.List; import java.util.Map; import java.util.Map.Entry; +import java.util.Objects; import java.util.Set; import java.util.TreeMap; import java.util.concurrent.ConcurrentHashMap; @@ -134,7 +134,7 @@ public Metrics(MetricConfig defaultConfig, List reporters, Time this.sensors = new ConcurrentHashMap<>(); this.metrics = new ConcurrentHashMap<>(); this.childrenSensors = new ConcurrentHashMap<>(); - this.reporters = Utils.notNull(reporters); + this.reporters = Objects.requireNonNull(reporters); this.time = time; for (MetricsReporter reporter : reporters) reporter.init(new ArrayList()); @@ -313,7 +313,7 @@ public MetricConfig config() { * @return Return the sensor or null if no such sensor exists */ public Sensor getSensor(String name) { - return this.sensors.get(Utils.notNull(name)); + return this.sensors.get(Objects.requireNonNull(name)); } /** @@ -500,8 +500,8 @@ public void addMetric(MetricName metricName, MetricConfig config, Measurable mea */ public void addMetric(MetricName metricName, MetricConfig config, MetricValueProvider metricValueProvider) { KafkaMetric m = new KafkaMetric(new Object(), - Utils.notNull(metricName), - Utils.notNull(metricValueProvider), + Objects.requireNonNull(metricName), + Objects.requireNonNull(metricValueProvider), config == null ? this.config : config, time); registerMetric(m); @@ -545,7 +545,7 @@ public synchronized KafkaMetric removeMetric(MetricName metricName) { * Add a MetricReporter */ public synchronized void addReporter(MetricsReporter reporter) { - Utils.notNull(reporter).init(new ArrayList<>(metrics.values())); + Objects.requireNonNull(reporter).init(new ArrayList<>(metrics.values())); this.reporters.add(reporter); } diff --git a/clients/src/main/java/org/apache/kafka/common/metrics/Sensor.java b/clients/src/main/java/org/apache/kafka/common/metrics/Sensor.java index 1af9419bc757a..7d52d01f20640 100644 --- a/clients/src/main/java/org/apache/kafka/common/metrics/Sensor.java +++ b/clients/src/main/java/org/apache/kafka/common/metrics/Sensor.java @@ -19,7 +19,6 @@ import org.apache.kafka.common.MetricName; import org.apache.kafka.common.metrics.CompoundStat.NamedMeasurable; import org.apache.kafka.common.utils.Time; -import org.apache.kafka.common.utils.Utils; import java.util.ArrayList; import java.util.HashSet; @@ -28,6 +27,7 @@ import java.util.List; import java.util.Locale; import java.util.Map; +import java.util.Objects; import java.util.Set; import java.util.concurrent.TimeUnit; @@ -107,7 +107,7 @@ public boolean shouldRecord(final int configId) { long inactiveSensorExpirationTimeSeconds, RecordingLevel recordingLevel) { super(); this.registry = registry; - this.name = Utils.notNull(name); + this.name = Objects.requireNonNull(name); this.parents = parents == null ? new Sensor[0] : parents; this.metrics = new LinkedHashMap<>(); this.stats = new ArrayList<>(); @@ -238,7 +238,7 @@ public synchronized boolean add(CompoundStat stat, MetricConfig config) { if (hasExpired()) return false; - this.stats.add(Utils.notNull(stat)); + this.stats.add(Objects.requireNonNull(stat)); Object lock = metricLock(); for (NamedMeasurable m : stat.stats()) { final KafkaMetric metric = new KafkaMetric(lock, m.name(), m.stat(), config == null ? this.config : config, time); @@ -276,8 +276,8 @@ public synchronized boolean add(final MetricName metricName, final MeasurableSta } else { final KafkaMetric metric = new KafkaMetric( metricLock(), - Utils.notNull(metricName), - Utils.notNull(stat), + Objects.requireNonNull(metricName), + Objects.requireNonNull(stat), config == null ? this.config : config, time ); diff --git a/clients/src/main/java/org/apache/kafka/common/utils/Utils.java b/clients/src/main/java/org/apache/kafka/common/utils/Utils.java index dbaf165ea5df3..c7b70af60a164 100755 --- a/clients/src/main/java/org/apache/kafka/common/utils/Utils.java +++ b/clients/src/main/java/org/apache/kafka/common/utils/Utils.java @@ -287,20 +287,6 @@ public static byte[] copyArray(byte[] src) { return Arrays.copyOf(src, src.length); } - /** - * Check that the parameter t is not null - * - * @param t The object to check - * @return t if it isn't null - * @throws NullPointerException if t is null. - */ - public static T notNull(T t) { - if (t == null) - throw new NullPointerException(); - else - return t; - } - /** * Sleep for a bit * @param ms The duration of the sleep diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java index 320ce114c61e7..ffd744bc7df02 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java @@ -35,6 +35,7 @@ import java.util.HashMap; import java.util.HashSet; import java.util.Map; +import java.util.Objects; import java.util.Set; import java.util.concurrent.ExecutionException; @@ -107,7 +108,7 @@ public void makeReady(final Map topics) { final Set newTopics = new HashSet<>(); for (final String topicName : topicsNotReady) { - final InternalTopicConfig internalTopicConfig = Utils.notNull(topics.get(topicName)); + final InternalTopicConfig internalTopicConfig = Objects.requireNonNull(topics.get(topicName)); final Map topicConfig = internalTopicConfig.getProperties(defaultTopicConfigs, windowChangeLogAdditionalRetention); log.debug("Going to create topic {} with {} partitions and config {}.",