Denovo1998 commented on issue #23149: URL: https://github.com/apache/pulsar/issues/23149#issuecomment-2282637452
@dao-jun @coderzc @mattisonchao **If this is not a bug, and there is no problem using it this way. Then the delay time change of the delay message can be easily implemented!** @HarryFQG https://github.com/apache/pulsar/discussions/20255 @liuhuagui https://github.com/apache/pulsar/discussions/18317 @hoswey https://github.com/apache/pulsar/issues/10042 @jaggerwang https://github.com/apache/pulsar/issues/21173 **This is a simple code example. It is no longer necessary for redis or database to store messages that need to change the delay time, and then filter them on the consumer side.** ```java @Test public void testDelayedMessageDelayTimeChange() throws Exception { final String topic = "persistent://my-property/my-ns/testDelayedMessageDelayTimeChange"; final String TASK_ID_OF_DELAY_TIME_CHANGE = "__DEPRECATED_TASK_ID"; final String DELAY_CHANGE_DEPRECATED_MESSAGE_ID = "__DELAY_CHANGE_DEPRECATED_MESSAGE_ID"; final String DELAY_CHANGE_NEW_MESSAGE_ID = "__DELAY_CHANGE_NEW_MESSAGE_ID"; Map<String, MessageId> receivedMessageIdMap = new ConcurrentHashMap<>(); Map<String, MessageId> sendMessageMap = new HashMap<>(); // ============================================================================================================ Consumer<byte[]> consumer = pulsarClient.newConsumer().topic(topic) .subscriptionName("my-subscription") .subscriptionType(SubscriptionType.Shared) .messageListener((MessageListener<byte[]>) (c, receivedMessage) -> { MessageId newTaskMessageId = receivedMessage.getMessageId(); String taskId = receivedMessage.getKey(); if (receivedMessage.hasProperty(DELAY_CHANGE_DEPRECATED_MESSAGE_ID)) { String deprecatedMessage = receivedMessage.getProperty(DELAY_CHANGE_DEPRECATED_MESSAGE_ID); byte[] bytes = deprecatedMessage.getBytes(UTF_8); try { if (receivedMessage.hasProperty(DELAY_CHANGE_NEW_MESSAGE_ID)) { newTaskMessageId = MessageId.fromByteArray(receivedMessage. getProperty(DELAY_CHANGE_NEW_MESSAGE_ID).getBytes(UTF_8)); MessageId deprecatedMessageId = MessageId.fromByteArray(bytes); log.warn("Received deprecated message with taskId {} , oldMessageId {} , newMessageId {}", receivedMessage.getProperty(TASK_ID_OF_DELAY_TIME_CHANGE), deprecatedMessageId, newTaskMessageId); c.acknowledge(deprecatedMessageId); } else { MessageId deprecatedMessageId = MessageId.fromByteArray(bytes); log.warn("Received deprecated message with taskId {} , oldMessageId {} , newMessageId {}", receivedMessage.getProperty(TASK_ID_OF_DELAY_TIME_CHANGE), deprecatedMessageId, newTaskMessageId); c.acknowledge(deprecatedMessageId); receivedMessageIdMap.putIfAbsent(taskId, newTaskMessageId); c.acknowledge(newTaskMessageId); } } catch (IOException e) { log.error("Format error. " + e.getCause()); } } else { receivedMessageIdMap.putIfAbsent(taskId, newTaskMessageId); c.acknowledgeAsync(newTaskMessageId); } }) .subscribe(); admin.topicPolicies().setDelayedDeliveryPolicy(topic, DelayedDeliveryPolicies.builder() .active(true) .tickTime(100L) .maxDeliveryDelayInMillis(5000) .build()); // ============================================================================================================ Producer<byte[]> producer = pulsarClient.newProducer().topic(topic).enableBatching(false).create(); final int taskNum = 10; for (int i = 0; i < taskNum; i++) { final String taskId = "task" + i; producer.newMessage() .key(taskId) .value(("Delayed message with task" + i).getBytes()) .deliverAfter(3, TimeUnit.SECONDS) .sendAsync().thenAccept(messageId -> { sendMessageMap.put(taskId, messageId); }); } producer.flush(); // =========================================================================================================== String taskId = "task5"; MessageId oldMessageId = sendMessageMap.get(taskId); // 1.Turn task5 into a delay of 1 seconds. Trigger task5 in advance. // deliverAfter: 3s -> 1s producer.newMessage() .key(taskId) .value(("Delayed message with " + taskId).getBytes()) .deliverAfter(1, TimeUnit.SECONDS) .property(TASK_ID_OF_DELAY_TIME_CHANGE, taskId) .property(DELAY_CHANGE_DEPRECATED_MESSAGE_ID, new String(sendMessageMap.get(taskId).toByteArray())) .sendAsync().thenAccept(messageId -> { log.warn("[Delayed message change] TaskId: {}, Old MessageId: {}, New MessageId: {}", taskId, sendMessageMap.get(taskId), messageId); sendMessageMap.put(taskId, messageId); }); producer.flush(); // 2.If you want to extend the delay time. // deliverAfter: 3s -> 5s producer.newMessage() .key(taskId) .value(("Delayed message with " + taskId).getBytes()) .deliverAfter(5, TimeUnit.SECONDS) .sendAsync().thenAccept(messageId -> { log.warn("[Delayed message change] TaskId: {}, Old MessageId: {}, New MessageId: {}", taskId, sendMessageMap.get(taskId), messageId); sendMessageMap.put(taskId, messageId); // A special "messages with no business implications" is sent just to change the delay time. producer.newMessage() .key("DEPRECATED_" + taskId) .value(("DEPRECATED_" + taskId).getBytes(UTF_8)) .property(TASK_ID_OF_DELAY_TIME_CHANGE, taskId) .property(DELAY_CHANGE_DEPRECATED_MESSAGE_ID, new String(oldMessageId.toByteArray(), UTF_8)) .property(DELAY_CHANGE_NEW_MESSAGE_ID, new String(messageId.toByteArray(), UTF_8)) .sendAsync(); }); producer.flush(); try { Awaitility.await().atMost(Duration.ofSeconds(30)).until(() -> receivedMessageIdMap.containsKey(taskId) && receivedMessageIdMap.size() == 10 && !receivedMessageIdMap.get(taskId).equals(oldMessageId)); } finally { consumer.close(); } } ``` **Choose one of 1 and 2 below.** -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
