[fix][client] Make DeadLetterPolicy & KeySharedPolicy serializable (apache#23718)

Co-authored-by: anurag.reddy <anurag.reddy@flipkart.com>
(cherry picked from commit 14129e3)
(cherry picked from commit b9ce087)

Author: 2 people

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BatchMessageTest.java

@@ -990,8 +990,8 @@ public void testBatchMessageDispatchingAccordingToPermits() throws Exception {
 
         int numMsgs = 1000;
         int batchMessages = 10;
-        final String topicName = "persistent://prop/ns-abc/testRetrieveSequenceIdSpecify-" + UUID.randomUUID();
-        final String subscriptionName = "sub-1";
+        final String topicName = "persistent://prop/ns-abc/testBatchMessageDispatchingAccordingToPermits-" + UUID.randomUUID();
+        final String subscriptionName = "bmdap-sub-1";
 
         ConsumerImpl<byte[]> consumer1 = (ConsumerImpl<byte[]>) pulsarClient.newConsumer().topic(topicName)
                 .subscriptionName(subscriptionName).receiverQueueSize(10).subscriptionType(SubscriptionType.Shared)
@@ -1016,6 +1016,7 @@ public void testBatchMessageDispatchingAccordingToPermits() throws Exception {
 
         producer.close();
         consumer1.close();
+        consumer2.close();
     }
 
     @Test(dataProvider="testSubTypeAndEnableBatch")

pulsar-client-api/src/main/java/org/apache/pulsar/client/api/DeadLetterPolicy.java

@@ -18,6 +18,7 @@
  */
 package org.apache.pulsar.client.api;
 
+import java.io.Serializable;
 import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
@@ -36,7 +37,8 @@
 @AllArgsConstructor
 @InterfaceAudience.Public
 @InterfaceStability.Stable
-public class DeadLetterPolicy {
+public class DeadLetterPolicy implements Serializable {
+    private static final long serialVersionUID = 1L;
 
     /**
      * Maximum number of times that a message will be redelivered before being sent to the dead letter queue.

pulsar-client-api/src/main/java/org/apache/pulsar/client/api/KeySharedPolicy.java

@@ -18,6 +18,7 @@
  */
 package org.apache.pulsar.client.api;
 
+import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
@@ -29,7 +30,7 @@
  */
 @InterfaceAudience.Public
 @InterfaceStability.Stable
-public abstract class KeySharedPolicy {
+public abstract class KeySharedPolicy implements Serializable {
 
     protected KeySharedMode keySharedMode;
 
@@ -82,6 +83,7 @@ public int getHashRangeTotal() {
      * for message, the cursor will rewind.
      */
     public static class KeySharedPolicySticky extends KeySharedPolicy {
+        private static final long serialVersionUID = 1L;
 
         protected final List<Range> ranges;
 
@@ -129,6 +131,7 @@ public List<Range> getRanges() {
      * Auto split hash range key shared policy.
      */
     public static class KeySharedPolicyAutoSplit extends KeySharedPolicy {
+        private static final long serialVersionUID = 1L;
 
         KeySharedPolicyAutoSplit() {
             this.keySharedMode = KeySharedMode.AUTO_SPLIT;

pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Range.java

@@ -18,6 +18,7 @@
  */
 package org.apache.pulsar.client.api;
 
+import java.io.Serializable;
 import java.util.Objects;
 import org.apache.pulsar.common.classification.InterfaceAudience;
 import org.apache.pulsar.common.classification.InterfaceStability;
@@ -27,7 +28,8 @@
  */
 @InterfaceAudience.Public
 @InterfaceStability.Stable
-public class Range implements Comparable<Range> {
+public class Range implements Comparable<Range>, Serializable {
+    private static final long serialVersionUID = 1L;
 
     private final int start;
     private final int end;

pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ConsumerConfigurationData.java

@@ -359,7 +359,7 @@ public int getMaxPendingChuckedMessage() {
                     + "When specifying the dead letter policy while not specifying `ackTimeoutMillis`, you can set the"
                     + " ack timeout to 30000 millisecond."
     )
-    private transient DeadLetterPolicy deadLetterPolicy;
+    private DeadLetterPolicy deadLetterPolicy;
 
     private boolean retryEnable = false;
 
@@ -388,7 +388,7 @@ public int getMaxPendingChuckedMessage() {
     private boolean resetIncludeHead = false;
 
     @JsonIgnore
-    private transient KeySharedPolicy keySharedPolicy;
+    private KeySharedPolicy keySharedPolicy;
 
     private boolean batchIndexAckEnabled = false;
 

pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ReaderConfigurationData.java

@@ -144,7 +144,7 @@ public class ReaderConfigurationData<T> implements Serializable, Cloneable {
     )
     private boolean resetIncludeHead = false;
 
-    private transient List<Range> keyHashRanges;
+    private List<Range> keyHashRanges;
 
     private boolean poolMessages = false;
 

pulsar-client/src/test/java/org/apache/pulsar/client/impl/conf/ConsumerConfigurationDataTest.java

@@ -19,7 +19,18 @@
 package org.apache.pulsar.client.impl.conf;
 
 import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.util.Collections;
 import java.util.regex.Pattern;
+
+import lombok.Cleanup;
+import org.apache.pulsar.client.api.DeadLetterPolicy;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.testng.Assert;
 import org.testng.annotations.DataProvider;
 import org.testng.annotations.Test;
 
@@ -45,4 +56,43 @@ public void testTopicConsumerConfigurationData(String topicName, int expectedPri
 
         assertThat(topicConsumerConfigurationData.getPriorityLevel()).isEqualTo(expectedPriority);
     }
+
+    @Test
+    public void testSerializable() throws Exception {
+        ConsumerConfigurationData<String> consumerConfigurationData = new ConsumerConfigurationData<>();
+        consumerConfigurationData.setPriorityLevel(1);
+        consumerConfigurationData.setSubscriptionName("my-sub");
+        consumerConfigurationData.setSubscriptionType(SubscriptionType.Shared);
+        consumerConfigurationData.setReceiverQueueSize(100);
+        consumerConfigurationData.setAckTimeoutMillis(1000);
+        consumerConfigurationData.setTopicNames(Collections.singleton("my-topic"));
+
+        DeadLetterPolicy deadLetterPolicy = DeadLetterPolicy.builder()
+                .maxRedeliverCount(10)
+                .retryLetterTopic("retry-topic")
+                .deadLetterTopic("dead-topic")
+                .build();
+        consumerConfigurationData.setDeadLetterPolicy(deadLetterPolicy);
+
+        @Cleanup
+        ByteArrayOutputStream bos = new ByteArrayOutputStream();
+        @Cleanup
+        ObjectOutputStream oos = new ObjectOutputStream(bos);
+        oos.writeObject(consumerConfigurationData);
+        byte[] serialized = bos.toByteArray();
+
+        // Deserialize
+        @Cleanup
+        ByteArrayInputStream bis = new ByteArrayInputStream(serialized);
+        @Cleanup
+        ObjectInputStream ois = new ObjectInputStream(bis);
+        Object object = ois.readObject();
+
+        Assert.assertEquals(object.getClass(), ConsumerConfigurationData.class);
+        Assert.assertEquals(object, consumerConfigurationData);
+
+        DeadLetterPolicy deserialisedDeadLetterPolicy = ((ConsumerConfigurationData<?>) object).getDeadLetterPolicy();
+        Assert.assertNotNull(deserialisedDeadLetterPolicy);
+        Assert.assertEquals(deserialisedDeadLetterPolicy, deadLetterPolicy);
+    }
 }