@Override
    public void onWebSocketText(String message) {
        if (log.isDebugEnabled()) {
            log.debug("[{}] Received new message from producer {} ", producer.getTopic(),
                    getRemote().getInetSocketAddress().toString());
        }
        ProducerMessage sendRequest;
        byte[] rawPayload = null;
        String requestContext = null;
        try {
            sendRequest = producerMessageReader.readValue(message);
            requestContext = sendRequest.context;
            rawPayload = Base64.getDecoder().decode(sendRequest.payload);
        } catch (IOException e) {
            sendAckResponse(new ProducerAck(FailedToDeserializeFromJSON, e.getMessage(), null, null));
            return;
        } catch (IllegalArgumentException e) {
            String msg = format("Invalid Base64 message-payload error=%s", e.getMessage());
            sendAckResponse(new ProducerAck(PayloadEncodingError, msg, null, requestContext));
            return;
        } catch (NullPointerException e) {
            // Null payload
            sendAckResponse(new ProducerAck(PayloadEncodingError, e.getMessage(), null, requestContext));
            return;
        }

        final long msgSize = rawPayload.length;
        TypedMessageBuilderImpl<byte[]> builder = (TypedMessageBuilderImpl<byte[]>) producer.newMessage();

        try {
            builder.value(rawPayload);
        } catch (SchemaSerializationException e) {
            sendAckResponse(new ProducerAck(PayloadEncodingError, e.getMessage(), null, requestContext));
            return;
        }

        if (sendRequest.properties != null) {
            builder.properties(sendRequest.properties);
        }
        if (sendRequest.key != null) {
            builder.key(sendRequest.key);
        }
        if (sendRequest.replicationClusters != null) {
            builder.replicationClusters(sendRequest.replicationClusters);
        }
        if (sendRequest.eventTime != null) {
            try {
                builder.eventTime(DateFormatter.parse(sendRequest.eventTime));
            } catch (DateTimeParseException e) {
                sendAckResponse(new ProducerAck(PayloadEncodingError, e.getMessage(), null, requestContext));
                return;
            }
        }
        if (sendRequest.deliverAt > 0) {
            builder.deliverAt(sendRequest.deliverAt);
        }
        if (sendRequest.deliverAfterMs > 0) {
            builder.deliverAfter(sendRequest.deliverAfterMs, TimeUnit.MILLISECONDS);
        }

        // If client-side encryption is enabled, the attributes "encryptParam", "uncompressedMessageSize",
        // "uncompressedMessageSize" and "batchSize" of message metadata must be set according to the parameters
        // when the client sends messages.
        if (clientSideEncrypt) {
            try {
                if (!StringUtils.isBlank(sendRequest.encryptionParam)) {
                    builder.getMetadataBuilder().setEncryptionParam(Base64.getDecoder()
                            .decode(sendRequest.encryptionParam));
                }
            } catch (Exception e){
                String msg = format("Invalid Base64 encryptionParam error=%s", e.getMessage());
                sendAckResponse(new ProducerAck(PayloadEncodingError, msg, null, requestContext));
                return;
            }
            if (sendRequest.compressionType != null && sendRequest.uncompressedMessageSize != null) {
                // Set compression information.
                builder.getMetadataBuilder().setCompression(sendRequest.compressionType);
                builder.getMetadataBuilder().setUncompressedSize(sendRequest.uncompressedMessageSize);
            } else if ((org.apache.pulsar.common.api.proto.CompressionType.NONE.equals(sendRequest.compressionType)
                    || sendRequest.compressionType == null)
                    && sendRequest.uncompressedMessageSize == null) {
                // Nothing to do, the method send async will set these two attributes.
            } else {
                // Only one param is set.
                sendAckResponse(new ProducerAck(PayloadEncodingError, "the params compressionType and"
                        + " uncompressedMessageSize should both empty or both non-empty",
                        null, requestContext));
                return;
            }
        }

        final long now = System.nanoTime();

        builder.sendAsync().thenAccept(msgId -> {
            if (log.isDebugEnabled()) {
                log.debug("[{}] Success fully write the message to broker with returned message ID {} from producer {}",
                        producer.getTopic(), msgId, getRemote().getInetSocketAddress().toString());
            }
            updateSentMsgStats(msgSize, TimeUnit.NANOSECONDS.toMicros(System.nanoTime() - now));
            if (isConnected()) {
                String messageId = Base64.getEncoder().encodeToString(msgId.toByteArray());
                sendAckResponse(new ProducerAck(messageId, sendRequest.context));
            }
        }).exceptionally(exception -> {
            log.warn("[{}] Error occurred while producer handler was sending msg from {}", producer.getTopic(),
                    getRemote().getInetSocketAddress().toString(), exception);
            numMsgsFailed.increment();
            sendAckResponse(
                    new ProducerAck(UnknownError, exception.getMessage(), null, sendRequest.context));
            return null;
        });
    }
@Test
    public void testProduceMessageAttributes() throws IOException {
        String producerV2 = "/ws/v2/producer/persistent/my-property/my-ns/my-topic";
        HttpServletRequest httpServletRequest = mock(HttpServletRequest.class);
        PulsarClient pulsarClient = mock(PulsarClient.class);
        ProducerBuilder producerBuilder = mock(ProducerBuilder.class);
        Producer producer = mock(Producer.class);
        TypedMessageBuilder messageBuilder = mock(TypedMessageBuilderImpl.class);
        ProducerMessage produceRequest = new ProducerMessage();

        produceRequest.setDeliverAfterMs(11111);
        produceRequest.setDeliverAt(22222);
        produceRequest.setContext("context");
        produceRequest.setPayload(Base64.getEncoder().encodeToString("my payload".getBytes()));

        // the params are all different with the default value
        Map<String, String[]> queryParams = new HashMap<>();

        httpServletRequest = mock(HttpServletRequest.class);
        when(httpServletRequest.getRequestURI()).thenReturn(producerV2);
        when(httpServletRequest.getParameterMap()).thenReturn(queryParams);

        WebSocketService service = mock(WebSocketService.class);
        when(service.isAuthenticationEnabled()).thenReturn(false);
        when(service.isAuthorizationEnabled()).thenReturn(false);
        when(service.getPulsarClient()).thenReturn(pulsarClient);

        when(pulsarClient.newProducer()).thenReturn(producerBuilder);
        when(producerBuilder.enableBatching(anyBoolean())).thenReturn(producerBuilder);
        when(producerBuilder.messageRoutingMode(any())).thenReturn(producerBuilder);
        when(producerBuilder.blockIfQueueFull(anyBoolean())).thenReturn(producerBuilder);
        when(producerBuilder.topic(anyString())).thenReturn(producerBuilder);
        when(producerBuilder.create()).thenReturn(producer);

        when(producer.newMessage()).thenReturn(messageBuilder);
        when(messageBuilder.sendAsync()).thenReturn( CompletableFuture.completedFuture(new MessageIdImpl(1, 2, 3)));

        ServletUpgradeResponse response = mock(ServletUpgradeResponse.class);

        ProducerHandler producerHandler = new ProducerHandler(service, httpServletRequest, response);
        producerHandler.onWebSocketText(ObjectMapperFactory.getMapper().writer().writeValueAsString(produceRequest));

        verify(messageBuilder, times(1)).deliverAfter(11111, TimeUnit.MILLISECONDS);
        verify(messageBuilder, times(1)).deliverAt(22222);
    }