MINOR: Substitute assertEquals(null) with assertNull (#9852)

Reviewers: David Jacot <djacot@confluent.io>

clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java

@@ -1067,7 +1067,7 @@ public class KafkaAdminClientTest {
             DescribeTopicsResult result = env.adminClient().describeTopics(Collections.singleton(topic));
             Map<String, TopicDescription> topicDescriptions = result.all().get();
             assertEquals(leader, topicDescriptions.get(topic).partitions().get(0).leader());
-            assertEquals(null, topicDescriptions.get(topic).authorizedOperations());
+            assertNull(topicDescriptions.get(topic).authorizedOperations());
         }
     }
 
@@ -1247,9 +1247,9 @@ public class KafkaAdminClientTest {
             DeleteAclsResult results = env.adminClient().deleteAcls(asList(FILTER1, FILTER2));
             Map<AclBindingFilter, KafkaFuture<FilterResults>> filterResults = results.values();
             FilterResults filter1Results = filterResults.get(FILTER1).get();
-            assertEquals(null, filter1Results.values().get(0).exception());
+            assertNull(filter1Results.values().get(0).exception());
             assertEquals(ACL1, filter1Results.values().get(0).binding());
-            assertEquals(null, filter1Results.values().get(1).exception());
+            assertNull(filter1Results.values().get(1).exception());
             assertEquals(ACL2, filter1Results.values().get(1).binding());
             TestUtils.assertFutureError(filterResults.get(FILTER2), SecurityDisabledException.class);
             TestUtils.assertFutureError(results.all(), SecurityDisabledException.class);
@@ -2068,7 +2068,7 @@ public class KafkaAdminClientTest {
             final DescribeClusterResult result = env.adminClient().describeCluster();
             assertEquals(env.cluster().clusterResource().clusterId(), result.clusterId().get());
             assertEquals(2, result.controller().get().id());
-            assertEquals(null, result.authorizedOperations().get());
+            assertNull(result.authorizedOperations().get());
 
             // Test DescribeCluster with the authorized operations included.
             final DescribeClusterResult result2 = env.adminClient().describeCluster();

clients/src/test/java/org/apache/kafka/clients/producer/internals/ProducerBatchTest.java

@@ -263,7 +263,7 @@ public class ProducerBatchTest {
         assertTrue(memoryRecordsBuilder.hasRoomFor(now, null, new byte[10], Record.EMPTY_HEADERS));
         memoryRecordsBuilder.closeForRecordAppends();
         assertFalse(memoryRecordsBuilder.hasRoomFor(now, null, new byte[10], Record.EMPTY_HEADERS));
-        assertEquals(null, batch.tryAppend(now + 1, null, new byte[10], Record.EMPTY_HEADERS, null, now + 1));
+        assertNull(batch.tryAppend(now + 1, null, new byte[10], Record.EMPTY_HEADERS, null, now + 1));
     }
 
     private static class MockCallback implements Callback {

clients/src/test/java/org/apache/kafka/common/KafkaFutureTest.java

@@ -29,6 +29,7 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 
 import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertEquals;
@@ -81,7 +82,7 @@ public class KafkaFutureTest {
         assertFalse(future.isCompletedExceptionally());
         assertFalse(future.isCancelled());
         myThread.join();
-        assertEquals(null, myThread.testException);
+        assertNull(myThread.testException);
     }
 
     @Test
@@ -183,8 +184,8 @@ public class KafkaFutureTest {
         for (int i = 0; i < numThreads; i++) {
             completerThreads.get(i).join();
             waiterThreads.get(i).join();
-            assertEquals(null, completerThreads.get(i).testException);
+            assertNull(completerThreads.get(i).testException);
-            assertEquals(null, waiterThreads.get(i).testException);
+            assertNull(waiterThreads.get(i).testException);
         }
     }
 

clients/src/test/java/org/apache/kafka/common/config/ConfigDefTest.java

@@ -95,7 +95,7 @@ public class ConfigDefTest {
         ConfigDef def = new ConfigDef().define("a", Type.INT, null, null, null, "docs");
         Map<String, Object> vals = def.parse(new Properties());
 
-        assertEquals(null, vals.get("a"));
+        assertNull(vals.get("a"));
     }
 
     @Test

clients/src/test/java/org/apache/kafka/common/config/provider/DirectoryConfigProviderTest.java

@@ -129,21 +129,21 @@ public class DirectoryConfigProviderTest {
     public void testEmptyPathWithKey() {
         ConfigData configData = provider.get("", Collections.singleton("foo"));
         assertTrue(configData.data().isEmpty());
-        assertEquals(null, configData.ttl());
+        assertNull(configData.ttl());
     }
 
     @Test
     public void testNullPath() {
         ConfigData configData = provider.get(null);
         assertTrue(configData.data().isEmpty());
-        assertEquals(null, configData.ttl());
+        assertNull(configData.ttl());
     }
 
     @Test
     public void testNullPathWithKey() {
         ConfigData configData = provider.get(null, Collections.singleton("foo"));
         assertTrue(configData.data().isEmpty());
-        assertEquals(null, configData.ttl());
+        assertNull(configData.ttl());
     }
 }
 

clients/src/test/java/org/apache/kafka/common/config/provider/FileConfigProviderTest.java

@@ -28,6 +28,7 @@ import java.util.HashMap;
 import java.util.Map;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
 public class FileConfigProviderTest {
@@ -46,7 +47,7 @@ public class FileConfigProviderTest {
         result.put("testKey", "testResult");
         result.put("testKey2", "testResult2");
         assertEquals(result, configData.data());
-        assertEquals(null, configData.ttl());
+        assertNull(configData.ttl());
     }
 
     @Test
@@ -55,35 +56,35 @@ public class FileConfigProviderTest {
         Map<String, String> result = new HashMap<>();
         result.put("testKey", "testResult");
         assertEquals(result, configData.data());
-        assertEquals(null, configData.ttl());
+        assertNull(configData.ttl());
     }
 
     @Test
     public void testEmptyPath() throws Exception {
         ConfigData configData = configProvider.get("", Collections.singleton("testKey"));
         assertTrue(configData.data().isEmpty());
-        assertEquals(null, configData.ttl());
+        assertNull(configData.ttl());
     }
 
     @Test
     public void testEmptyPathWithKey() throws Exception {
         ConfigData configData = configProvider.get("");
         assertTrue(configData.data().isEmpty());
-        assertEquals(null, configData.ttl());
+        assertNull(configData.ttl());
     }
 
     @Test
     public void testNullPath() throws Exception {
         ConfigData configData = configProvider.get(null);
         assertTrue(configData.data().isEmpty());
-        assertEquals(null, configData.ttl());
+        assertNull(configData.ttl());
     }
 
     @Test
     public void testNullPathWithKey() throws Exception {
         ConfigData configData = configProvider.get(null, Collections.singleton("testKey"));
         assertTrue(configData.data().isEmpty());
-        assertEquals(null, configData.ttl());
+        assertNull(configData.ttl());
     }
 
     public static class TestFileConfigProvider extends FileConfigProvider {

clients/src/test/java/org/apache/kafka/common/protocol/MessageUtilTest.java

@@ -29,6 +29,7 @@ import java.util.Collections;
 import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
 public final class MessageUtilTest {
@@ -53,7 +54,7 @@ public final class MessageUtilTest {
 
     @Test
     public void testDuplicate() {
-        assertEquals(null, MessageUtil.duplicate(null));
+        assertNull(MessageUtil.duplicate(null));
         assertArrayEquals(new byte[] {},
             MessageUtil.duplicate(new byte[] {}));
         assertArrayEquals(new byte[] {1, 2, 3},

clients/src/test/java/org/apache/kafka/common/protocol/types/ProtocolSerializationTest.java

@@ -29,6 +29,7 @@ import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.fail;
 
@@ -399,7 +400,7 @@ public class ProtocolSerializationTest {
         Struct newFormat = newSchema.read(buffer);
         assertEquals(value, newFormat.get("field1"));
         assertEquals("default", newFormat.get("field2"));
-        assertEquals(null, newFormat.get("field3"));
+        assertNull(newFormat.get("field3"));
         assertEquals(ByteBuffer.allocate(0), newFormat.get("field4"));
         assertEquals(Long.MAX_VALUE, newFormat.get("field5"));
     }

clients/src/test/java/org/apache/kafka/common/utils/CircularIteratorTest.java

@@ -18,6 +18,7 @@
 package org.apache.kafka.common.utils;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 
@@ -48,9 +49,9 @@ public class CircularIteratorTest {
         assertEquals("B", it.peek());
         assertTrue(it.hasNext());
         assertEquals("B", it.next());
-        assertEquals(null, it.peek());
+        assertNull(it.peek());
         assertTrue(it.hasNext());
-        assertEquals(null, it.next());
+        assertNull(it.next());
         assertEquals("C", it.peek());
         assertTrue(it.hasNext());
         assertEquals("C", it.next());

clients/src/test/java/org/apache/kafka/common/utils/ConfigUtilsTest.java

@@ -45,15 +45,15 @@ public class ConfigUtilsTest {
             {"cow", "beef", "heifer", "steer"}
         });
         assertEquals("baz", newConfig.get("foo.bar"));
-        assertEquals(null, newConfig.get("foobar.deprecated"));
+        assertNull(newConfig.get("foobar.deprecated"));
         assertEquals("1", newConfig.get("chicken"));
-        assertEquals(null, newConfig.get("rooster"));
+        assertNull(newConfig.get("rooster"));
-        assertEquals(null, newConfig.get("hen"));
+        assertNull(newConfig.get("hen"));
         assertEquals("moo", newConfig.get("cow"));
-        assertEquals(null, newConfig.get("beef"));
+        assertNull(newConfig.get("beef"));
-        assertEquals(null, newConfig.get("heifer"));
+        assertNull(newConfig.get("heifer"));
-        assertEquals(null, newConfig.get("steer"));
+        assertNull(newConfig.get("steer"));
-        assertEquals(null, config.get("cow"));
+        assertNull(config.get("cow"));
         assertEquals("blah", config.get("blah"));
         assertEquals("blah", newConfig.get("blah"));
         assertEquals(42, newConfig.get("unexpected.non.string.object"));

connect/api/src/test/java/org/apache/kafka/connect/data/SchemaProjectorTest.java

@@ -30,6 +30,7 @@ import java.util.List;
 import java.util.Map;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.fail;
 
@@ -493,7 +494,7 @@ public class SchemaProjectorTest {
     public void testProjectMissingOptionalStructField() {
         final Schema source = SchemaBuilder.struct().build();
         final Schema target = SchemaBuilder.struct().field("id", SchemaBuilder.OPTIONAL_INT64_SCHEMA).build();
-        assertEquals(null, ((Struct) SchemaProjector.project(source, new Struct(source), target)).getInt64("id"));
+        assertNull(((Struct) SchemaProjector.project(source, new Struct(source), target)).getInt64("id"));
     }
 
     @Test
@@ -521,7 +522,7 @@ public class SchemaProjectorTest {
 
         projected = SchemaProjector.project(source, null, target);
         if (optional) {
-            assertEquals(null, projected);
+            assertNull(projected);
         } else {
             assertEquals(defaultValue, projected);
         }

connect/api/src/test/java/org/apache/kafka/connect/storage/StringConverterTest.java

@@ -25,6 +25,7 @@ import java.util.Collections;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
 
 public class StringConverterTest {
     private static final String TOPIC = "topic";
@@ -44,7 +45,7 @@ public class StringConverterTest {
 
     @Test
     public void testNullToBytes() {
-        assertEquals(null, converter.fromConnectData(TOPIC, Schema.OPTIONAL_STRING_SCHEMA, null));
+        assertNull(converter.fromConnectData(TOPIC, Schema.OPTIONAL_STRING_SCHEMA, null));
     }
 
     @Test
@@ -69,7 +70,7 @@ public class StringConverterTest {
     public void testBytesNullToString() {
         SchemaAndValue data = converter.toConnectData(TOPIC, null);
         assertEquals(Schema.OPTIONAL_STRING_SCHEMA, data.schema());
-        assertEquals(null, data.value());
+        assertNull(data.value());
     }
 
     @Test
@@ -95,6 +96,6 @@ public class StringConverterTest {
 
     @Test
     public void testNullHeaderValueToBytes() {
-        assertEquals(null, converter.fromConnectHeader(TOPIC, "hdr", Schema.OPTIONAL_STRING_SCHEMA, null));
+        assertNull(converter.fromConnectHeader(TOPIC, "hdr", Schema.OPTIONAL_STRING_SCHEMA, null));
     }
 }

connect/file/src/test/java/org/apache/kafka/connect/file/FileStreamSourceTaskTest.java

@@ -37,6 +37,7 @@ import java.util.List;
 import java.util.Map;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
 
 public class FileStreamSourceTaskTest extends EasyMockSupport {
 
@@ -84,10 +85,10 @@ public class FileStreamSourceTaskTest extends EasyMockSupport {
         task.start(config);
 
         OutputStream os = Files.newOutputStream(tempFile.toPath());
-        assertEquals(null, task.poll());
+        assertNull(task.poll());
         os.write("partial line".getBytes());
         os.flush();
-        assertEquals(null, task.poll());
+        assertNull(task.poll());
         os.write(" finished\n".getBytes());
         os.flush();
         List<SourceRecord> records = task.poll();
@@ -96,7 +97,7 @@ public class FileStreamSourceTaskTest extends EasyMockSupport {
         assertEquals("partial line finished", records.get(0).value());
         assertEquals(Collections.singletonMap(FileStreamSourceTask.FILENAME_FIELD, tempFile.getAbsolutePath()), records.get(0).sourcePartition());
         assertEquals(Collections.singletonMap(FileStreamSourceTask.POSITION_FIELD, 22L), records.get(0).sourceOffset());
-        assertEquals(null, task.poll());
+        assertNull(task.poll());
 
         // Different line endings, and make sure the final \r doesn't result in a line until we can
         // read the subsequent byte.
@@ -224,7 +225,7 @@ public class FileStreamSourceTaskTest extends EasyMockSupport {
         task.start(config);
         // Currently the task retries indefinitely if the file isn't found, but shouldn't return any data.
         for (int i = 0; i < 100; i++)
-            assertEquals(null, task.poll());
+            assertNull(task.poll());
     }
 
 

connect/runtime/src/test/java/org/apache/kafka/connect/converters/NumberConverterTest.java

@@ -95,7 +95,7 @@ public abstract class NumberConverterTest<T extends Number> {
 
     @Test
     public void testNullToBytes() {
-        assertEquals(null, converter.fromConnectData(TOPIC, schema, null));
+        assertNull(converter.fromConnectData(TOPIC, schema, null));
     }
 
     @Test

connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskTest.java

@@ -1349,7 +1349,7 @@ public class WorkerSinkTaskTest {
         SinkRecord record = records.getValue().iterator().next();
 
         // we expect null for missing timestamp, the sentinel value of Record.NO_TIMESTAMP is Kafka's API
-        assertEquals(null, record.timestamp());
+        assertNull(record.timestamp());
         assertEquals(TimestampType.CREATE_TIME, record.timestampType());
 
         PowerMock.verifyAll();

connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskTest.java

@@ -579,7 +579,7 @@ public class WorkerSourceTaskTest extends ThreadedTest {
 
         Whitebox.setInternalState(workerTask, "toSend", records);
         Whitebox.invokeMethod(workerTask, "sendRecords");
-        assertEquals(null, sent.getValue().timestamp());
+        assertNull(sent.getValue().timestamp());
 
         PowerMock.verifyAll();
     }

connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSourceTaskWithTopicCreationTest.java

@@ -603,7 +603,7 @@ public class WorkerSourceTaskWithTopicCreationTest extends ThreadedTest {
 
         Whitebox.setInternalState(workerTask, "toSend", records);
         Whitebox.invokeMethod(workerTask, "sendRecords");
-        assertEquals(null, sent.getValue().timestamp());
+        assertNull(sent.getValue().timestamp());
 
         PowerMock.verifyAll();
     }

connect/runtime/src/test/java/org/apache/kafka/connect/storage/FileOffsetBackingStoreTest.java

@@ -35,6 +35,7 @@ import java.util.Map;
 import java.util.concurrent.ThreadPoolExecutor;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
 @RunWith(PowerMockRunner.class)
@@ -84,7 +85,7 @@ public class FileOffsetBackingStoreTest {
 
         Map<ByteBuffer, ByteBuffer> values = store.get(Arrays.asList(buffer("key"), buffer("bad"))).get();
         assertEquals(buffer("value"), values.get(buffer("key")));
-        assertEquals(null, values.get(buffer("bad")));
+        assertNull(values.get(buffer("bad")));
 
         PowerMock.verifyAll();
     }

connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaStatusBackingStoreFormatTest.java

@@ -50,6 +50,7 @@ import static org.easymock.EasyMock.expectLastCall;
 import static org.easymock.EasyMock.newCapture;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
 @RunWith(PowerMockRunner.class)
@@ -198,7 +199,7 @@ public class KafkaStatusBackingStoreFormatTest extends EasyMockSupport {
         // check capture state
         assertEquals(topicStatus, store.parseTopicStatus(valueCapture.getValue()));
         // state is not visible until read back from the log
-        assertEquals(null, store.getTopic(FOO_CONNECTOR, FOO_TOPIC));
+        assertNull(store.getTopic(FOO_CONNECTOR, FOO_TOPIC));
 
         ConsumerRecord<String, byte[]> statusRecord = new ConsumerRecord<>(STATUS_TOPIC, 0, 0, key, valueCapture.getValue());
         store.read(statusRecord);
@@ -231,7 +232,7 @@ public class KafkaStatusBackingStoreFormatTest extends EasyMockSupport {
         // check capture state
         assertEquals(topicStatus, store.parseTopicStatus(valueCapture.getValue()));
         // state is not visible until read back from the log
-        assertEquals(null, store.getTopic(FOO_CONNECTOR, FOO_TOPIC));
+        assertNull(store.getTopic(FOO_CONNECTOR, FOO_TOPIC));
 
         verifyAll();
     }
@@ -257,7 +258,7 @@ public class KafkaStatusBackingStoreFormatTest extends EasyMockSupport {
         // check capture state
         assertEquals(topicStatus, store.parseTopicStatus(valueCapture.getValue()));
         // state is not visible until read back from the log
-        assertEquals(null, store.getTopic(FOO_CONNECTOR, FOO_TOPIC));
+        assertNull(store.getTopic(FOO_CONNECTOR, FOO_TOPIC));
 
         verifyAll();
     }

connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaStatusBackingStoreTest.java

@@ -54,6 +54,7 @@ import static org.easymock.EasyMock.expect;
 import static org.easymock.EasyMock.expectLastCall;
 import static org.easymock.EasyMock.newCapture;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 
@@ -114,7 +115,7 @@ public class KafkaStatusBackingStoreTest extends EasyMockSupport {
         store.put(status);
 
         // state is not visible until read back from the log
-        assertEquals(null, store.get(CONNECTOR));
+        assertNull(store.get(CONNECTOR));
 
         verifyAll();
     }
@@ -148,7 +149,7 @@ public class KafkaStatusBackingStoreTest extends EasyMockSupport {
         store.put(status);
 
         // state is not visible until read back from the log
-        assertEquals(null, store.get(CONNECTOR));
+        assertNull(store.get(CONNECTOR));
 
         verifyAll();
     }
@@ -176,7 +177,7 @@ public class KafkaStatusBackingStoreTest extends EasyMockSupport {
         store.put(status);
 
         // state is not visible until read back from the log
-        assertEquals(null, store.get(CONNECTOR));
+        assertNull(store.get(CONNECTOR));
 
         verifyAll();
     }
@@ -365,7 +366,7 @@ public class KafkaStatusBackingStoreTest extends EasyMockSupport {
         store.put(status);
 
         // state is not visible until read back from the log
-        assertEquals(null, store.get(TASK));
+        assertNull(store.get(TASK));
 
         verifyAll();
     }

connect/runtime/src/test/java/org/apache/kafka/connect/util/ConnectUtilsTest.java

@@ -97,7 +97,7 @@ public class ConnectUtilsTest {
 
         Map<String, Object> prop = new HashMap<>();
         ConnectUtils.addMetricsContextProperties(prop, config, "cluster-1");
-        assertEquals(null, prop.get(CommonClientConfigs.METRICS_CONTEXT_PREFIX + WorkerConfig.CONNECT_GROUP_ID));
+        assertNull(prop.get(CommonClientConfigs.METRICS_CONTEXT_PREFIX + WorkerConfig.CONNECT_GROUP_ID));
         assertEquals("cluster-1", prop.get(CommonClientConfigs.METRICS_CONTEXT_PREFIX + WorkerConfig.CONNECT_KAFKA_CLUSTER_ID));
 
     }

connect/runtime/src/test/java/org/apache/kafka/connect/util/LoggingContextTest.java

@@ -203,6 +203,6 @@ public class LoggingContextTest {
     }
 
     protected void assertConnectorMdcUnset() {
-        assertEquals(null, MDC.get(EXTRA_KEY3));
+        assertNull(MDC.get(EXTRA_KEY3));
     }
 }

connect/transforms/src/test/java/org/apache/kafka/connect/transforms/FlattenTest.java

@@ -310,8 +310,8 @@ public class FlattenTest {
                 null, null);
         final SourceRecord transformedRecord = xformValue.apply(record);
 
-        assertEquals(null, transformedRecord.value());
+        assertNull(transformedRecord.value());
-        assertEquals(null, transformedRecord.valueSchema());
+        assertNull(transformedRecord.valueSchema());
     }
 
     @Test
@@ -323,7 +323,7 @@ public class FlattenTest {
                 simpleStructSchema, null);
         final SourceRecord transformedRecord = xformValue.apply(record);
 
-        assertEquals(null, transformedRecord.value());
+        assertNull(transformedRecord.value());
         assertEquals(simpleStructSchema, transformedRecord.valueSchema());
     }
 }

connect/transforms/src/test/java/org/apache/kafka/connect/transforms/InsertFieldTest.java

@@ -31,6 +31,7 @@ import java.util.HashMap;
 import java.util.Map;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertThrows;
 
@@ -131,8 +132,8 @@ public class InsertFieldTest {
 
         final SourceRecord transformedRecord = xformValue.apply(record);
 
-        assertEquals(null, transformedRecord.value());
+        assertNull(transformedRecord.value());
-        assertEquals(null, transformedRecord.valueSchema());
+        assertNull(transformedRecord.valueSchema());
     }
 
     @Test
@@ -153,7 +154,7 @@ public class InsertFieldTest {
 
         final SourceRecord transformedRecord = xformValue.apply(record);
 
-        assertEquals(null, transformedRecord.value());
+        assertNull(transformedRecord.value());
         assertEquals(simpleStructSchema, transformedRecord.valueSchema());
     }
 
@@ -176,9 +177,9 @@ public class InsertFieldTest {
         assertEquals(42L, ((Map<?, ?>) transformedRecord.key()).get("magic"));
         assertEquals("test", ((Map<?, ?>) transformedRecord.key()).get("topic_field"));
         assertEquals(0, ((Map<?, ?>) transformedRecord.key()).get("partition_field"));
-        assertEquals(null, ((Map<?, ?>) transformedRecord.key()).get("timestamp_field"));
+        assertNull(((Map<?, ?>) transformedRecord.key()).get("timestamp_field"));
         assertEquals("my-instance-id", ((Map<?, ?>) transformedRecord.key()).get("instance_id"));
-        assertEquals(null, transformedRecord.value());
+        assertNull(transformedRecord.value());
     }
 
     @Test

connect/transforms/src/test/java/org/apache/kafka/connect/transforms/SetSchemaMetadataTest.java

@@ -29,6 +29,7 @@ import java.util.HashMap;
 import java.util.Map;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertSame;
 
 public class SetSchemaMetadataTest {
@@ -133,7 +134,7 @@ public class SetSchemaMetadataTest {
     @Test
     public void updateSchemaOfNull() {
         Object updatedValue = SetSchemaMetadata.updateSchemaIn(null, Schema.INT32_SCHEMA);
-        assertEquals(null, updatedValue);
+        assertNull(updatedValue);
     }
 
     protected void assertMatchingSchema(Struct value, Schema schema) {

core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala

@@ -1238,7 +1238,7 @@ class PlaintextConsumerTest extends BaseConsumerTest {
     assertEquals(80, timestampTopic3P0.timestamp)
     assertEquals(Optional.of(0), timestampTopic3P0.leaderEpoch)
 
-    assertEquals(null, timestampOffsets.get(new TopicPartition(topic3, 1)))
+    assertNull(timestampOffsets.get(new TopicPartition(topic3, 1)))
   }
 
   @Test

core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorTest.scala

@@ -1388,7 +1388,7 @@ class GroupCoordinatorTest {
     timer.advanceClock(DefaultRebalanceTimeout + 1)
     // The static leader should already session timeout, moving group towards Empty
     assertEquals(Set.empty, getGroup(groupId).allMembers)
-    assertEquals(null, getGroup(groupId).leaderOrNull)
+    assertNull(getGroup(groupId).leaderOrNull)
     assertEquals(3, getGroup(groupId).generationId)
     assertGroupState(groupState = Empty)
   }

core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala

@@ -389,7 +389,7 @@ class KafkaConfigTest {
     val conf2 = KafkaConfig.fromProps(props)
     assertEquals(listenerListToEndPoints("PLAINTEXT://:1111"), conf2.listeners)
     assertEquals(listenerListToEndPoints("PLAINTEXT://:1111"), conf2.advertisedListeners)
-    assertEquals(null, conf2.listeners.find(_.securityProtocol == SecurityProtocol.PLAINTEXT).get.host)
+    assertNull(conf2.listeners.find(_.securityProtocol == SecurityProtocol.PLAINTEXT).get.host)
 
     // configuration with advertised host and port, and no advertised listeners
     props.put(KafkaConfig.AdvertisedHostNameProp, "otherhost")

streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java

@@ -90,7 +90,7 @@ import java.util.Set;
  * assertEquals("one", driver.flushedEntryStored(1));
  * assertEquals("two", driver.flushedEntryStored(2));
  * assertEquals("four", driver.flushedEntryStored(4));
- * assertEquals(null, driver.flushedEntryStored(5));
+ * assertNull(driver.flushedEntryStored(5));
  *
  * assertEquals(false, driver.flushedEntryRemoved(0));
  * assertEquals(false, driver.flushedEntryRemoved(1));