[consumer] Fixed NPE in ChangelogClientConfig (#1683)

Author: FelixGV

clients/da-vinci-client/src/main/java/com/linkedin/davinci/consumer/ChangelogClientConfig.java

@@ -4,12 +4,14 @@
 import com.linkedin.venice.client.store.ClientConfig;
 import com.linkedin.venice.controllerapi.D2ControllerClient;
 import com.linkedin.venice.schema.SchemaReader;
+import java.util.Objects;
 import java.util.Properties;
+import javax.annotation.Nonnull;
 import org.apache.avro.specific.SpecificRecord;
 
 
 public class ChangelogClientConfig<T extends SpecificRecord> {
-  private Properties consumerProperties;
+  private @Nonnull Properties consumerProperties = new Properties();
   private SchemaReader schemaReader;
   private String viewName;
   private Boolean isBeforeImageView = false;
@@ -67,11 +69,12 @@ public String getStoreName() {
     return innerClientConfig.getStoreName();
   }
 
-  public ChangelogClientConfig<T> setConsumerProperties(Properties consumerProperties) {
-    this.consumerProperties = consumerProperties;
+  public ChangelogClientConfig<T> setConsumerProperties(@Nonnull Properties consumerProperties) {
+    this.consumerProperties = Objects.requireNonNull(consumerProperties);
     return this;
   }
 
+  @Nonnull
   public Properties getConsumerProperties() {
     return consumerProperties;
   }

clients/da-vinci-client/src/test/java/com/linkedin/davinci/consumer/VeniceChangelogConsumerImplTest.java

@@ -8,6 +8,10 @@
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertThrows;
+import static org.testng.Assert.assertTrue;
 
 import com.linkedin.avroutil1.compatibility.AvroCompatibilityHelper;
 import com.linkedin.davinci.consumer.stats.BasicConsumerStats;
@@ -65,6 +69,7 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Properties;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
@@ -104,6 +109,19 @@ public void setUp() {
     valueSerializer = FastSerializerDeserializerFactory.getFastAvroGenericSerializer(valueSchema);
   }
 
+  @Test
+  public void testConfig() {
+    ChangelogClientConfig config = new ChangelogClientConfig();
+    assertNotNull(config.getConsumerProperties());
+    assertTrue(config.getConsumerProperties().isEmpty());
+    assertThrows(NullPointerException.class, () -> config.setConsumerProperties(null));
+    Properties newProps = new Properties();
+    newProps.setProperty("foo", "bar");
+    config.setConsumerProperties(newProps);
+    assertNotNull(config.getConsumerProperties());
+    assertFalse(config.getConsumerProperties().isEmpty());
+  }
+
   @Test
   public void testConsumeBeforeAndAfterImage() throws ExecutionException, InterruptedException {
     D2ControllerClient d2ControllerClient = mock(D2ControllerClient.class);
@@ -394,7 +412,7 @@ public void testBootstrapState() {
         0);
     doReturn(currentTimestamp).when(veniceChangelogConsumer).getSubscribeTime();
     veniceChangelogConsumer.maybeUpdatePartitionToBootstrapMap(message, pubSubTopicPartition);
-    Assert.assertFalse(bootstrapStateMap.get(0));
+    assertFalse(bootstrapStateMap.get(0));
     kafkaMessageEnvelope.producerMetadata.messageTimestamp = currentTimestamp - TimeUnit.SECONDS.toMillis(30);
     veniceChangelogConsumer.maybeUpdatePartitionToBootstrapMap(message, pubSubTopicPartition);
     Assert.assertTrue(bootstrapStateMap.get(0));
@@ -453,7 +471,7 @@ public void testConsumeAfterImage() throws ExecutionException, InterruptedExcept
     prepareVersionTopicRecordsToBePolled(5L, 15L, mockPubSubConsumer, oldVersionTopic, 0, true);
     pubSubMessages =
         (List<PubSubMessage<String, ChangeEvent<Utf8>, VeniceChangeCoordinate>>) veniceChangelogConsumer.poll(100);
-    Assert.assertFalse(pubSubMessages.isEmpty());
+    assertFalse(pubSubMessages.isEmpty());
     Assert.assertEquals(pubSubMessages.size(), 10);
     for (int i = 5; i < 15; i++) {
       PubSubMessage<String, ChangeEvent<Utf8>, VeniceChangeCoordinate> pubSubMessage = pubSubMessages.get(i - 5);
@@ -526,7 +544,7 @@ public void testConsumeAfterImageWithCompaction() throws ExecutionException, Int
 
     prepareVersionTopicRecordsToBePolled(5L, 15L, mockPubSubConsumer, oldVersionTopic, 0, true);
     pubSubMessages = new ArrayList<>(veniceChangelogConsumer.poll(100));
-    Assert.assertFalse(pubSubMessages.isEmpty());
+    assertFalse(pubSubMessages.isEmpty());
     Assert.assertEquals(pubSubMessages.size(), 10);
     for (int i = 5; i < 15; i++) {
       PubSubMessage<String, ChangeEvent<Utf8>, VeniceChangeCoordinate> pubSubMessage = pubSubMessages.get(i - 5);

gradle/spotbugs/exclude.xml

@@ -75,6 +75,7 @@
     <Or>
       <Class name="com.linkedin.venice.router.api.TestHostFinder"/>
       <Class name="com.linkedin.davinci.StoreBackendTest"/>
+      <Class name="com.linkedin.davinci.consumer.VeniceChangelogConsumerImplTest"/>
       <Class name="com.linkedin.venice.memory.ClassSizeEstimatorTest"/>
       <Class name="com.linkedin.venice.controller.server.VeniceControllerAccessManagerTest"/>
       <Class name="com.linkedin.venice.sql.SQLUtilsTest"/>