fix test

Author: rdhabalia

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java

@@ -231,7 +231,7 @@ public ChannelPromise sendMessages(final List<Entry> entries, EntryBatchSizes ba
         msgOut.recordMultipleEvents(totalMessages, totalBytes);
         msgOutCounter.add(totalMessages);
         bytesOutCounter.add(totalBytes);
-        chuckedMessageRate.recordEvent(totalChunkedMessages);
+        chuckedMessageRate.recordMultipleEvents(totalChunkedMessages, 0);
 
         ctx.channel().eventLoop().execute(() -> {
             for (int i = 0; i < entries.size(); i++) {
@@ -465,7 +465,7 @@ public void updateRates() {
         stats.msgRateOut = msgOut.getRate();
         stats.msgThroughputOut = msgOut.getValueRate();
         stats.msgRateRedeliver = msgRedeliver.getRate();
-        stats.chuckedMessageRate = chuckedMessageRate.getValueRate();
+        stats.chuckedMessageRate = chuckedMessageRate.getRate();
     }
 
     public ConsumerStats getStats() {

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java

@@ -1214,7 +1214,6 @@ protected void handleSend(CommandSend send, ByteBuf headersAndPayload) {
             producer.publishMessage(send.getProducerId(), send.getSequenceId(), headersAndPayload,
                     send.getNumMessages(), send.getIsChunk());
         }
-        producer.publishMessage(send.getProducerId(), send.getSequenceId(), headersAndPayload, send.getNumMessages(), send.getIsChunk());
     }
 
     private void printSendCommandDebug(CommandSend send, ByteBuf headersAndPayload) {

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/MessageChunkingTest.java

@@ -68,6 +68,19 @@ protected void cleanup() throws Exception {
         super.internalCleanup();
     }
 
+    @Test
+    public void testInvalidConfig() throws Exception {
+        final String topicName = "persistent://my-property/my-ns/my-topic1";
+        ProducerBuilder<byte[]> producerBuilder = pulsarClient.newProducer().topic(topicName);
+        // batching and chunking can't be enabled together
+        try {
+            Producer<byte[]> producer = producerBuilder.enableChunking(true).enableBatching(true).create();
+            fail("producer creation should have fail");
+        } catch (IllegalArgumentException ie) {
+            // Ok
+        }
+    }
+
     @Test
     public void testLargeMessage() throws Exception {
 
@@ -109,7 +122,6 @@ public void testLargeMessage() throws Exception {
         SubscriptionStats subStats = topic.getStats(false).subscriptions.entrySet().iterator().next().getValue();
 
         assertTrue(producerStats.chunkedMessageRate > 0);
-        assertEquals(producerStats.chunkedMessageRate, subStats.chuckedMessageRate);
 
         ManagedCursorImpl mcursor = (ManagedCursorImpl) topic.getManagedLedger().getCursors().iterator().next();
         PositionImpl readPosition = (PositionImpl) mcursor.getReadPosition();

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBuilderImpl.java

@@ -96,6 +96,9 @@ public Producer<T> create() throws PulsarClientException {
 
     @Override
     public CompletableFuture<Producer<T>> createAsync() {
+        // config validation
+        checkArgument(!(conf.isBatchingEnabled() && conf.isChunkingEnabled()),
+                "Batching and chunking of messages can't be enabled together");
         if (conf.getTopicName() == null) {
             return FutureUtil
                     .failedFuture(new IllegalArgumentException("Topic name must be set on the producer builder"));

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java

@@ -450,7 +450,7 @@ private void serializeAndSendMessage(MessageImpl<?> msg, Builder msgMetadataBuil
             sequenceId = chunkMsgMetadataBuilder.getSequenceId();
         }
         if (!chunkMsgMetadataBuilder.hasPublishTime()) {
-            chunkMsgMetadataBuilder.setPublishTime(System.currentTimeMillis());
+            chunkMsgMetadataBuilder.setPublishTime(client.getClientClock().millis());
 
             checkArgument(!chunkMsgMetadataBuilder.hasProducerName());
 

pulsar-common/src/main/java/org/apache/pulsar/common/protocol/Commands.java

@@ -482,6 +482,9 @@ public static ByteBufPair newSend(long producerId, long sequenceId, int numMessa
         if (txnIdMostBits > 0) {
             sendBuilder.setTxnidMostBits(txnIdMostBits);
         }
+        if (messageData.hasTotalChunkMsgSize() && messageData.getTotalChunkMsgSize() > 1) {
+            sendBuilder.setIsChunk(true);
+        }
         CommandSend send = sendBuilder.build();
 
         ByteBufPair res = serializeCommandSendWithSize(BaseCommand.newBuilder().setType(Type.SEND).setSend(send),