You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@eventmesh.apache.org by "HarshSawarkar (via GitHub)" <gi...@apache.org> on 2024/02/12 15:13:55 UTC
[PR] Implemented retry strategy based on pulsar message middleware (eventmesh)
HarshSawarkar opened a new pull request, #4769:
URL: https://github.com/apache/eventmesh/pull/4769
<!--
### Contribution Checklist
- Name the pull request in the form "[ISSUE #XXXX] Title of the pull request",
where *XXXX* should be replaced by the actual issue number.
Skip *[ISSUE #XXXX]* if there is no associated github issue for this pull request.
- Fill out the template below to describe the changes contributed by the pull request.
That will give reviewers the context they need to do the review.
- Each pull request should address only one issue.
Please do not mix up code from multiple issues.
- Each commit in the pull request should have a meaningful commit message.
- Once all items of the checklist are addressed, remove the above text and this checklist,
leaving only the filled out template below.
(The sections below can be removed for hotfixes of typos)
-->
<!--
(If this PR fixes a GitHub issue, please add `Fixes #<XXX>` or `Closes #<XXX>`.)
-->
Fixes #4556
### Motivation
Earlier the retry strategy based on pulsar middleware were only supported
### Modifications
Implemented retry strategy based on the Pulsar message middleware.
### Documentation
- Does this pull request introduce a new feature? (yes / no)
- If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)
- If a feature is not applicable for documentation, explain why?
- If a feature is not documented yet in this PR, please create a followup issue for adding the documentation
--
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: dev-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: dev-help@eventmesh.apache.org
Re: [PR] [ISSUE #4556]Implemented retry strategy based on pulsar message middleware (eventmesh)
Posted by "HarshSawarkar (via GitHub)" <gi...@apache.org>.
HarshSawarkar commented on code in PR #4769:
URL: https://github.com/apache/eventmesh/pull/4769#discussion_r1502985258
##########
eventmesh-retry/eventmesh-retry-pulsar/src/main/java/org/apache/eventmesh/retry/pulsar/PulsarRetryStrategyImpl.java:
##########
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.retry.pulsar;
+
+import org.apache.eventmesh.api.SendCallback;
+import org.apache.eventmesh.api.SendResult;
+import org.apache.eventmesh.api.exception.OnExceptionContext;
+import org.apache.eventmesh.api.producer.Producer;
+import org.apache.eventmesh.common.protocol.http.common.ProtocolKey;
+import org.apache.eventmesh.common.protocol.http.common.ProtocolKey.ClientInstanceKey;
+import org.apache.eventmesh.retry.api.conf.RetryConfiguration;
+import org.apache.eventmesh.retry.api.strategy.RetryStrategy;
+
+import org.apache.pulsar.client.util.RetryMessageUtil;
+
+import java.util.Objects;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class PulsarRetryStrategyImpl implements RetryStrategy {
+ @Override
+ public void retry(final RetryConfiguration configuration) {
+ sendMessageBack(configuration);
+ }
+
+ @SneakyThrows
+ private void sendMessageBack(final RetryConfiguration configuration) {
+ CloudEvent event = configuration.getEvent();
+ String topic = configuration.getTopic();
+ String consumerGroupName = configuration.getConsumerGroupName();
+ String retryTopicName = consumerGroupName + RetryMessageUtil.RETRY_GROUP_TOPIC_SUFFIX;
Review Comment:
Could you provide some implementation ideas on how to implement the feature for automatically retrying failed messages of Pulsar?
--
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: issues-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [PR] [ISSUE #4556] Implement retry strategy based on pulsar message middleware (eventmesh)
Posted by "pandaapo (via GitHub)" <gi...@apache.org>.
pandaapo commented on code in PR #4769:
URL: https://github.com/apache/eventmesh/pull/4769#discussion_r1567215145
##########
eventmesh-retry/eventmesh-retry-pulsar/src/main/java/org/apache/eventmesh/retry/pulsar/PulsarRetryStrategyImpl.java:
##########
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.retry.pulsar;
+
+import org.apache.eventmesh.api.SendCallback;
+import org.apache.eventmesh.api.SendResult;
+import org.apache.eventmesh.api.exception.OnExceptionContext;
+import org.apache.eventmesh.api.producer.Producer;
+import org.apache.eventmesh.common.protocol.http.common.ProtocolKey;
+import org.apache.eventmesh.common.protocol.http.common.ProtocolKey.ClientInstanceKey;
+import org.apache.eventmesh.retry.api.conf.RetryConfiguration;
+import org.apache.eventmesh.retry.api.strategy.RetryStrategy;
+
+import org.apache.pulsar.client.util.RetryMessageUtil;
+
+import java.util.Objects;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class PulsarRetryStrategyImpl implements RetryStrategy {
+ @Override
+ public void retry(final RetryConfiguration configuration) {
+ sendMessageBack(configuration);
+ }
+
+ @SneakyThrows
+ private void sendMessageBack(final RetryConfiguration configuration) {
+ CloudEvent event = configuration.getEvent();
+ String topic = configuration.getTopic();
+ String consumerGroupName = configuration.getConsumerGroupName();
+ String retryTopicName = consumerGroupName + RetryMessageUtil.RETRY_GROUP_TOPIC_SUFFIX;
Review Comment:
> you'll need to resend the messages through EventMesh. You can achieve this using an in-memory queue and a timer.
This suggestion is incorrect. Because the task of this PR is to implement retry in Pulsar without relying on EventMesh's memory. Otherwise, the PR would be redundant since EventMesh retries within the memory by default.
--
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: issues-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [PR] [ISSUE #4556]Implemented retry strategy based on pulsar message middleware (eventmesh)
Posted by "pandaapo (via GitHub)" <gi...@apache.org>.
pandaapo commented on code in PR #4769:
URL: https://github.com/apache/eventmesh/pull/4769#discussion_r1497030681
##########
eventmesh-retry/eventmesh-retry-pulsar/src/main/java/org/apache/eventmesh/retry/pulsar/PulsarRetryStrategyImpl.java:
##########
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.retry.pulsar;
+
+import org.apache.eventmesh.api.SendCallback;
+import org.apache.eventmesh.api.SendResult;
+import org.apache.eventmesh.api.exception.OnExceptionContext;
+import org.apache.eventmesh.api.producer.Producer;
+import org.apache.eventmesh.common.protocol.http.common.ProtocolKey;
+import org.apache.eventmesh.common.protocol.http.common.ProtocolKey.ClientInstanceKey;
+import org.apache.eventmesh.retry.api.conf.RetryConfiguration;
+import org.apache.eventmesh.retry.api.strategy.RetryStrategy;
+
+import org.apache.pulsar.client.util.RetryMessageUtil;
+
+import java.util.Objects;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class PulsarRetryStrategyImpl implements RetryStrategy {
+ @Override
+ public void retry(final RetryConfiguration configuration) {
+ sendMessageBack(configuration);
+ }
+
+ @SneakyThrows
+ private void sendMessageBack(final RetryConfiguration configuration) {
+ CloudEvent event = configuration.getEvent();
+ String topic = configuration.getTopic();
+ String consumerGroupName = configuration.getConsumerGroupName();
+ String retryTopicName = consumerGroupName + RetryMessageUtil.RETRY_GROUP_TOPIC_SUFFIX;
Review Comment:
I am not familiar with Pulsar, but I have two questions about your code:
1 Pulsar's message retry feature is turned off by default and needs to be turned on manually, how did you turn it on?
2 I don't understand how you realize that Pulsar automatically retries messages in the "xxx-RETRY" queue. Could you explain?
@HarshSawarkar
--
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: issues-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [PR] [ISSUE #4556]Implemented retry strategy based on pulsar message middleware (eventmesh)
Posted by "HarshSawarkar (via GitHub)" <gi...@apache.org>.
HarshSawarkar commented on code in PR #4769:
URL: https://github.com/apache/eventmesh/pull/4769#discussion_r1501560839
##########
eventmesh-retry/eventmesh-retry-pulsar/src/main/java/org/apache/eventmesh/retry/pulsar/PulsarRetryStrategyImpl.java:
##########
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.retry.pulsar;
+
+import org.apache.eventmesh.api.SendCallback;
+import org.apache.eventmesh.api.SendResult;
+import org.apache.eventmesh.api.exception.OnExceptionContext;
+import org.apache.eventmesh.api.producer.Producer;
+import org.apache.eventmesh.common.protocol.http.common.ProtocolKey;
+import org.apache.eventmesh.common.protocol.http.common.ProtocolKey.ClientInstanceKey;
+import org.apache.eventmesh.retry.api.conf.RetryConfiguration;
+import org.apache.eventmesh.retry.api.strategy.RetryStrategy;
+
+import org.apache.pulsar.client.util.RetryMessageUtil;
+
+import java.util.Objects;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class PulsarRetryStrategyImpl implements RetryStrategy {
+ @Override
+ public void retry(final RetryConfiguration configuration) {
+ sendMessageBack(configuration);
+ }
+
+ @SneakyThrows
+ private void sendMessageBack(final RetryConfiguration configuration) {
+ CloudEvent event = configuration.getEvent();
+ String topic = configuration.getTopic();
+ String consumerGroupName = configuration.getConsumerGroupName();
+ String retryTopicName = consumerGroupName + RetryMessageUtil.RETRY_GROUP_TOPIC_SUFFIX;
Review Comment:
Hi @pandaapo, @xwm1992 the retry feature is not turned on. For turning it on, can the **_retryAsynchronously(Supplier<CompletableFuture<T>> supplier, Backoff backoff, ScheduledExecutorService scheduledExecutorService, CompletableFuture<T> callback)_** method of **_RetryUtil_** class of **_pulsar-client-2.11.1.jar_** library be considered as it provides with the method **_scheduledExecutorService.execute()_** to schedule and execute a retry.
As in **_eventmesh-storage-RocketMQ_** the class **_ConsumeMessageConcurrentlyService_** has the implemenetation for retrying failed message of **RocketMQ**. Can you please guide?
--
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: issues-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [PR] [ISSUE #4556]Implemented retry strategy based on pulsar message middleware (eventmesh)
Posted by "pandaapo (via GitHub)" <gi...@apache.org>.
pandaapo commented on PR #4769:
URL: https://github.com/apache/eventmesh/pull/4769#issuecomment-1956048381
This PR only implements a part of issue #4556, so please create a new issue and "Fixes #<it>".
--
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: issues-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [PR] [ISSUE #4556]Implemented retry strategy based on pulsar message middleware (eventmesh)
Posted by "HarshSawarkar (via GitHub)" <gi...@apache.org>.
HarshSawarkar commented on code in PR #4769:
URL: https://github.com/apache/eventmesh/pull/4769#discussion_r1501560839
##########
eventmesh-retry/eventmesh-retry-pulsar/src/main/java/org/apache/eventmesh/retry/pulsar/PulsarRetryStrategyImpl.java:
##########
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.retry.pulsar;
+
+import org.apache.eventmesh.api.SendCallback;
+import org.apache.eventmesh.api.SendResult;
+import org.apache.eventmesh.api.exception.OnExceptionContext;
+import org.apache.eventmesh.api.producer.Producer;
+import org.apache.eventmesh.common.protocol.http.common.ProtocolKey;
+import org.apache.eventmesh.common.protocol.http.common.ProtocolKey.ClientInstanceKey;
+import org.apache.eventmesh.retry.api.conf.RetryConfiguration;
+import org.apache.eventmesh.retry.api.strategy.RetryStrategy;
+
+import org.apache.pulsar.client.util.RetryMessageUtil;
+
+import java.util.Objects;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class PulsarRetryStrategyImpl implements RetryStrategy {
+ @Override
+ public void retry(final RetryConfiguration configuration) {
+ sendMessageBack(configuration);
+ }
+
+ @SneakyThrows
+ private void sendMessageBack(final RetryConfiguration configuration) {
+ CloudEvent event = configuration.getEvent();
+ String topic = configuration.getTopic();
+ String consumerGroupName = configuration.getConsumerGroupName();
+ String retryTopicName = consumerGroupName + RetryMessageUtil.RETRY_GROUP_TOPIC_SUFFIX;
Review Comment:
Hi @pandaapo, @xwm1992 the retry feature is not turned on. For turning it on can the **_retryAsynchronously(Supplier<CompletableFuture<T>> supplier, Backoff backoff, ScheduledExecutorService scheduledExecutorService, CompletableFuture<T> callback)_** method of **_RetryUtil_** class of **_pulsar-client-2.11.1.jar_** library be considered as it provides with the method **_scheduledExecutorService.execute()_** to schedule and execute a retry. Can you guide?
--
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: issues-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [PR] [ISSUE #4556]Implemented retry strategy based on pulsar message middleware (eventmesh)
Posted by "pandaapo (via GitHub)" <gi...@apache.org>.
pandaapo commented on code in PR #4769:
URL: https://github.com/apache/eventmesh/pull/4769#discussion_r1497030681
##########
eventmesh-retry/eventmesh-retry-pulsar/src/main/java/org/apache/eventmesh/retry/pulsar/PulsarRetryStrategyImpl.java:
##########
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.retry.pulsar;
+
+import org.apache.eventmesh.api.SendCallback;
+import org.apache.eventmesh.api.SendResult;
+import org.apache.eventmesh.api.exception.OnExceptionContext;
+import org.apache.eventmesh.api.producer.Producer;
+import org.apache.eventmesh.common.protocol.http.common.ProtocolKey;
+import org.apache.eventmesh.common.protocol.http.common.ProtocolKey.ClientInstanceKey;
+import org.apache.eventmesh.retry.api.conf.RetryConfiguration;
+import org.apache.eventmesh.retry.api.strategy.RetryStrategy;
+
+import org.apache.pulsar.client.util.RetryMessageUtil;
+
+import java.util.Objects;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class PulsarRetryStrategyImpl implements RetryStrategy {
+ @Override
+ public void retry(final RetryConfiguration configuration) {
+ sendMessageBack(configuration);
+ }
+
+ @SneakyThrows
+ private void sendMessageBack(final RetryConfiguration configuration) {
+ CloudEvent event = configuration.getEvent();
+ String topic = configuration.getTopic();
+ String consumerGroupName = configuration.getConsumerGroupName();
+ String retryTopicName = consumerGroupName + RetryMessageUtil.RETRY_GROUP_TOPIC_SUFFIX;
Review Comment:
I am not familiar with Pulsar, but I have two questions about your code:
1 Pulsar's message retry feature is turned off by default and needs to be turned on manually, how did you turn it on?
2 I don't understand how you realize that Pulsar automatically retries messages in the "xxx-RETRY" queue. Could you explain?
--
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: issues-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [PR] [ISSUE #4556]Implemented retry strategy based on pulsar message middleware (eventmesh)
Posted by "pandaapo (via GitHub)" <gi...@apache.org>.
pandaapo commented on code in PR #4769:
URL: https://github.com/apache/eventmesh/pull/4769#discussion_r1501993326
##########
eventmesh-retry/eventmesh-retry-pulsar/src/main/java/org/apache/eventmesh/retry/pulsar/PulsarRetryStrategyImpl.java:
##########
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.retry.pulsar;
+
+import org.apache.eventmesh.api.SendCallback;
+import org.apache.eventmesh.api.SendResult;
+import org.apache.eventmesh.api.exception.OnExceptionContext;
+import org.apache.eventmesh.api.producer.Producer;
+import org.apache.eventmesh.common.protocol.http.common.ProtocolKey;
+import org.apache.eventmesh.common.protocol.http.common.ProtocolKey.ClientInstanceKey;
+import org.apache.eventmesh.retry.api.conf.RetryConfiguration;
+import org.apache.eventmesh.retry.api.strategy.RetryStrategy;
+
+import org.apache.pulsar.client.util.RetryMessageUtil;
+
+import java.util.Objects;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class PulsarRetryStrategyImpl implements RetryStrategy {
+ @Override
+ public void retry(final RetryConfiguration configuration) {
+ sendMessageBack(configuration);
+ }
+
+ @SneakyThrows
+ private void sendMessageBack(final RetryConfiguration configuration) {
+ CloudEvent event = configuration.getEvent();
+ String topic = configuration.getTopic();
+ String consumerGroupName = configuration.getConsumerGroupName();
+ String retryTopicName = consumerGroupName + RetryMessageUtil.RETRY_GROUP_TOPIC_SUFFIX;
Review Comment:
@HarshSawarkar
RocketMQ can automatically retry messages in the "% RETRY%..."queue, so `RocketMQRetryStrategyImpl` utilizes RocketMQ's retry queue feature and sends failed messages to a queue with the prefix "% RETRY%".
When it comes to Pulsar, we also need Pulsar to automatically retry failed messages. Currently your code does not implement this feature.
--
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: issues-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [PR] Implemented retry strategy based on pulsar message middleware (eventmesh)
Posted by "HarshSawarkar (via GitHub)" <gi...@apache.org>.
HarshSawarkar closed pull request #4769: Implemented retry strategy based on pulsar message middleware
URL: https://github.com/apache/eventmesh/pull/4769
--
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: dev-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: dev-help@eventmesh.apache.org
Re: [PR] [ISSUE #4556] Implement retry strategy based on pulsar message middleware (eventmesh)
Posted by "Pil0tXia (via GitHub)" <gi...@apache.org>.
Pil0tXia commented on code in PR #4769:
URL: https://github.com/apache/eventmesh/pull/4769#discussion_r1565933238
##########
eventmesh-retry/eventmesh-retry-pulsar/src/main/java/org/apache/eventmesh/retry/pulsar/PulsarRetryStrategyImpl.java:
##########
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.retry.pulsar;
+
+import org.apache.eventmesh.api.SendCallback;
+import org.apache.eventmesh.api.SendResult;
+import org.apache.eventmesh.api.exception.OnExceptionContext;
+import org.apache.eventmesh.api.producer.Producer;
+import org.apache.eventmesh.common.protocol.http.common.ProtocolKey;
+import org.apache.eventmesh.common.protocol.http.common.ProtocolKey.ClientInstanceKey;
+import org.apache.eventmesh.retry.api.conf.RetryConfiguration;
+import org.apache.eventmesh.retry.api.strategy.RetryStrategy;
+
+import org.apache.pulsar.client.util.RetryMessageUtil;
+
+import java.util.Objects;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class PulsarRetryStrategyImpl implements RetryStrategy {
+ @Override
+ public void retry(final RetryConfiguration configuration) {
+ sendMessageBack(configuration);
+ }
+
+ @SneakyThrows
+ private void sendMessageBack(final RetryConfiguration configuration) {
+ CloudEvent event = configuration.getEvent();
+ String topic = configuration.getTopic();
+ String consumerGroupName = configuration.getConsumerGroupName();
+ String retryTopicName = consumerGroupName + RetryMessageUtil.RETRY_GROUP_TOPIC_SUFFIX;
Review Comment:
@HarshSawarkar This PR has hit a standstill. After reading the conversation, it appears that Pulsar doesn't have a built-in retry queue, so you'll need to resend the messages through EventMesh. You can achieve this using an in-memory queue and a timer.
--
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: issues-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [PR] [ISSUE #4556]Implemented retry strategy based on pulsar message middleware (eventmesh)
Posted by "pandaapo (via GitHub)" <gi...@apache.org>.
pandaapo commented on code in PR #4769:
URL: https://github.com/apache/eventmesh/pull/4769#discussion_r1506941102
##########
eventmesh-retry/eventmesh-retry-pulsar/src/main/java/org/apache/eventmesh/retry/pulsar/PulsarRetryStrategyImpl.java:
##########
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.retry.pulsar;
+
+import org.apache.eventmesh.api.SendCallback;
+import org.apache.eventmesh.api.SendResult;
+import org.apache.eventmesh.api.exception.OnExceptionContext;
+import org.apache.eventmesh.api.producer.Producer;
+import org.apache.eventmesh.common.protocol.http.common.ProtocolKey;
+import org.apache.eventmesh.common.protocol.http.common.ProtocolKey.ClientInstanceKey;
+import org.apache.eventmesh.retry.api.conf.RetryConfiguration;
+import org.apache.eventmesh.retry.api.strategy.RetryStrategy;
+
+import org.apache.pulsar.client.util.RetryMessageUtil;
+
+import java.util.Objects;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class PulsarRetryStrategyImpl implements RetryStrategy {
+ @Override
+ public void retry(final RetryConfiguration configuration) {
+ sendMessageBack(configuration);
+ }
+
+ @SneakyThrows
+ private void sendMessageBack(final RetryConfiguration configuration) {
+ CloudEvent event = configuration.getEvent();
+ String topic = configuration.getTopic();
+ String consumerGroupName = configuration.getConsumerGroupName();
+ String retryTopicName = consumerGroupName + RetryMessageUtil.RETRY_GROUP_TOPIC_SUFFIX;
Review Comment:
@HarshSawarkar I wish I could give more useful help, but as I said earlier, I'm not familiar with Pulsar.
--
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: issues-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [PR] [ISSUE #4556] Implement retry strategy based on pulsar message middleware (eventmesh)
Posted by "HarshSawarkar (via GitHub)" <gi...@apache.org>.
HarshSawarkar closed pull request #4769: [ISSUE #4556] Implement retry strategy based on pulsar message middleware
URL: https://github.com/apache/eventmesh/pull/4769
--
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: dev-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: dev-help@eventmesh.apache.org
Re: [PR] [ISSUE #4556]Implemented retry strategy based on pulsar message middleware (eventmesh)
Posted by "HarshSawarkar (via GitHub)" <gi...@apache.org>.
HarshSawarkar commented on code in PR #4769:
URL: https://github.com/apache/eventmesh/pull/4769#discussion_r1501560839
##########
eventmesh-retry/eventmesh-retry-pulsar/src/main/java/org/apache/eventmesh/retry/pulsar/PulsarRetryStrategyImpl.java:
##########
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.retry.pulsar;
+
+import org.apache.eventmesh.api.SendCallback;
+import org.apache.eventmesh.api.SendResult;
+import org.apache.eventmesh.api.exception.OnExceptionContext;
+import org.apache.eventmesh.api.producer.Producer;
+import org.apache.eventmesh.common.protocol.http.common.ProtocolKey;
+import org.apache.eventmesh.common.protocol.http.common.ProtocolKey.ClientInstanceKey;
+import org.apache.eventmesh.retry.api.conf.RetryConfiguration;
+import org.apache.eventmesh.retry.api.strategy.RetryStrategy;
+
+import org.apache.pulsar.client.util.RetryMessageUtil;
+
+import java.util.Objects;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class PulsarRetryStrategyImpl implements RetryStrategy {
+ @Override
+ public void retry(final RetryConfiguration configuration) {
+ sendMessageBack(configuration);
+ }
+
+ @SneakyThrows
+ private void sendMessageBack(final RetryConfiguration configuration) {
+ CloudEvent event = configuration.getEvent();
+ String topic = configuration.getTopic();
+ String consumerGroupName = configuration.getConsumerGroupName();
+ String retryTopicName = consumerGroupName + RetryMessageUtil.RETRY_GROUP_TOPIC_SUFFIX;
Review Comment:
Hi @pandaapo the retry feature is not turned on. For turning it on can the **_retryAsynchronously(Supplier<CompletableFuture<T>> supplier, Backoff backoff, ScheduledExecutorService scheduledExecutorService, CompletableFuture<T> callback)_** method of **_RetryUtil_** class of **_pulsar-client-2.11.1.jar_** library be considered as it provides with the method **_scheduledExecutorService.execute()_** to schedule and execute a retry. Can you guide?
--
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: issues-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [PR] [ISSUE #4556]Implemented retry strategy based on pulsar message middleware (eventmesh)
Posted by "HarshSawarkar (via GitHub)" <gi...@apache.org>.
HarshSawarkar commented on code in PR #4769:
URL: https://github.com/apache/eventmesh/pull/4769#discussion_r1501560839
##########
eventmesh-retry/eventmesh-retry-pulsar/src/main/java/org/apache/eventmesh/retry/pulsar/PulsarRetryStrategyImpl.java:
##########
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.retry.pulsar;
+
+import org.apache.eventmesh.api.SendCallback;
+import org.apache.eventmesh.api.SendResult;
+import org.apache.eventmesh.api.exception.OnExceptionContext;
+import org.apache.eventmesh.api.producer.Producer;
+import org.apache.eventmesh.common.protocol.http.common.ProtocolKey;
+import org.apache.eventmesh.common.protocol.http.common.ProtocolKey.ClientInstanceKey;
+import org.apache.eventmesh.retry.api.conf.RetryConfiguration;
+import org.apache.eventmesh.retry.api.strategy.RetryStrategy;
+
+import org.apache.pulsar.client.util.RetryMessageUtil;
+
+import java.util.Objects;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class PulsarRetryStrategyImpl implements RetryStrategy {
+ @Override
+ public void retry(final RetryConfiguration configuration) {
+ sendMessageBack(configuration);
+ }
+
+ @SneakyThrows
+ private void sendMessageBack(final RetryConfiguration configuration) {
+ CloudEvent event = configuration.getEvent();
+ String topic = configuration.getTopic();
+ String consumerGroupName = configuration.getConsumerGroupName();
+ String retryTopicName = consumerGroupName + RetryMessageUtil.RETRY_GROUP_TOPIC_SUFFIX;
Review Comment:
Hi @pandaapo, @xwm1992 the retry feature is not turned on. For turning it on, can the **_retryAsynchronously(Supplier<CompletableFuture<T>> supplier, Backoff backoff, ScheduledExecutorService scheduledExecutorService, CompletableFuture<T> callback)_** method of **_RetryUtil_** class of **_pulsar-client-2.11.1.jar_** library be considered as it provides with the method **_scheduledExecutorService.execute()_** to schedule and execute a retry.
As in **_eventmesh-storage-RocketMQ_** the class **_ConsumeMessageConcurrentlyService_** has the implemenetation for retrying failed message. Can you please guide?
--
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: issues-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org