Fix pulsar entry formatter encode zero timestamp record caused exception (#1839)

### Motivation

When using pulsar entry formatter and publishing zero timestamp record
will cause request timeout,
the root cause is we will set the event time when timestamp >= 0, but
the event time has a check to ensure time > 0.
See:
```java
    public PulsarMessageBuilder eventTime(long timestamp) {
        checkArgument(timestamp > 0, "Invalid timestamp : '%s'", timestamp);
        metadata.setEventTime(timestamp);
        return this;
    }
```

### Modifications
* Do not check the event time before setting the event time.
* Add test to verify this issue.

(cherry picked from commit c7cb1cf)

Author: Demogorgon314

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/format/PulsarEntryFormatter.java

@@ -154,7 +154,7 @@ private static MessageImpl<ByteBuffer> recordToEntry(Record record) {
 
         // timestamp
         if (record.timestamp() >= 0) {
-            builder.eventTime(record.timestamp());
+            builder.getMetadataBuilder().setEventTime(record.timestamp());
             builder.getMetadataBuilder().setPublishTime(record.timestamp());
         } else {
             builder.getMetadataBuilder().setPublishTime(System.currentTimeMillis());

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/PulsarMessageBuilder.java

@@ -93,12 +93,6 @@ public PulsarMessageBuilder properties(Map<String, String> properties) {
         return this;
     }
 
-    public PulsarMessageBuilder eventTime(long timestamp) {
-        checkArgument(timestamp > 0, "Invalid timestamp : '%s'", timestamp);
-        metadata.setEventTime(timestamp);
-        return this;
-    }
-
     public MessageMetadata getMetadataBuilder() {
         return metadata;
     }
@@ -113,4 +107,4 @@ public Message<ByteBuffer> getMessage() {
         return MessageImpl.create(metadata, content, SCHEMA, null);
     }
 
-}
+}

tests/src/test/java/io/streamnative/pulsar/handlers/kop/BasicEndToEndPulsarTest.java

@@ -20,16 +20,20 @@
 import io.streamnative.pulsar.handlers.kop.utils.KopTopic;
 import java.time.Duration;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Properties;
 import java.util.TreeMap;
 import java.util.concurrent.CompletableFuture;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import lombok.Cleanup;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.pulsar.broker.service.Topic;
 import org.apache.pulsar.client.api.Consumer;
 import org.apache.pulsar.client.api.Producer;
@@ -136,4 +140,32 @@ public void testSkipReplicatedSubscriptionsMarker() throws Exception {
         kafkaConsumer.commitSync(Duration.ofSeconds(1));
         kafkaConsumer.close();
     }
+
+    @Test(timeOut = 30000)
+    public void testPublishZeroTimestampRecord() {
+        String topic = "test-publish-zero-timestamp-record";
+        String subscription = "test-group";
+        Properties properties = newKafkaProducerProperties();
+        @Cleanup
+        final KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
+        producer.send(new ProducerRecord<>(
+                topic,
+                null,
+                0L,
+                "k1",
+                "v1",
+                null)
+        );
+
+        producer.flush();
+
+        @Cleanup
+        final KafkaConsumer<String, String> consumer = newKafkaConsumer(topic, subscription);
+        consumer.subscribe(Collections.singleton(topic));
+        List<ConsumerRecord<String, String>> consumerRecords = receiveRecords(consumer, 1);
+        assertEquals(consumerRecords.size(), 1);
+        assertEquals(consumerRecords.get(0).key(), "k1");
+        assertEquals(consumerRecords.get(0).value(), "v1");
+        assertEquals(consumerRecords.get(0).timestamp(), 0L);
+    }
 }