You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@rocketmq.apache.org by GitBox <gi...@apache.org> on 2022/03/14 11:32:14 UTC
[GitHub] [rocketmq] dongeforever commented on a change in pull request #3987: [RIP-37] Add new APIs for producer
dongeforever commented on a change in pull request #3987:
URL: https://github.com/apache/rocketmq/pull/3987#discussion_r825838652
##########
File path: apis/src/main/java/org/apache/rocketmq/apis/producer/Producer.java
##########
@@ -0,0 +1,132 @@
+/*
+ * 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.rocketmq.apis.producer;
+
+import com.google.common.util.concurrent.Service;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.io.Closeable;
+import java.util.Collection;
+import org.apache.rocketmq.apis.exception.AuthenticationException;
+import org.apache.rocketmq.apis.exception.AuthorisationException;
+import org.apache.rocketmq.apis.exception.ClientException;
+import org.apache.rocketmq.apis.exception.MessageTypeDoesNotMatchException;
+import org.apache.rocketmq.apis.exception.PersistenceException;
+import org.apache.rocketmq.apis.exception.ProducerClosedAlreadyException;
+import org.apache.rocketmq.apis.exception.NetworkConnectionException;
+import org.apache.rocketmq.apis.exception.NetworkTimeoutException;
+import org.apache.rocketmq.apis.exception.TopicDoesNotExistException;
+import org.apache.rocketmq.apis.exception.TransactionCheckerNotSetException;
+import org.apache.rocketmq.apis.message.Message;
+import org.apache.rocketmq.apis.message.MessageId;
+
+/**
+ * Producer is a thread-safe rocketmq client which is used to publish messages.
+ *
+ * <p>On account of network timeout or other reasons, rocketmq producer only promised the at-least-once semantics.
+ * For producer, at-least-once semantics means potentially attempts are made at sending it, messages may be
+ * duplicated but not lost.
+ */
+public interface Producer extends Closeable {
+ /**
+ * Sends a message synchronously.
+ *
+ * <p>This method does not return until it gets the definitive result.
+ *
+ * @param message message to send.
+ * @return the message id assigned to the appointed message.
+ * @throws TopicDoesNotExistException if the topic of message does not exist.
+ * @throws AuthorisationException if no permission to send message.
+ * @throws AuthenticationException if identification could not be recognized by server.
+ * @throws ProducerClosedAlreadyException if producer is closed already.
+ * @throws MessageTypeDoesNotMatchException if message type does not match with the topic.
+ * @throws NetworkTimeoutException if encountered network timeout to communicate with server.
+ * @throws NetworkConnectionException if there is a network connection problem.
+ * @throws PersistenceException if encountered persistence failure from server.
+ */
+ MessageId send(Message message) throws ClientException;
+
+ /**
+ * Sends a transactional message synchronously.
+ *
+ * @param message message to send.
+ * @param transaction transaction to bind.
+ * @return the message id assigned to the appointed message.
+ * @throws TopicDoesNotExistException if the topic of message does not exist.
+ * @throws AuthorisationException if no permission to send message.
+ * @throws AuthenticationException if identification could not be recognized by server.
+ * @throws ProducerClosedAlreadyException if producer is closed already.
+ * @throws MessageTypeDoesNotMatchException if message type does not match with the topic.
+ * @throws NetworkTimeoutException if encountered network timeout to communicate with server.
+ * @throws NetworkConnectionException if there is a network connection problem.
+ * @throws PersistenceException if encountered persistence failure from server.
+ * @throws TransactionCheckerNotSetException if {@link TransactionChecker} is not set.
+ */
+ MessageId send(Message message, Transaction transaction) throws ClientException;
+
+ /**
+ * Send a message asynchronously.
+ *
+ * <p>This method returns immediately, the result is included in the {@link CompletableFuture};
+ *
+ * @param message message to send.
+ * @return a future that indicates the result.
+ */
+ CompletableFuture<MessageId> sendAsync(Message message);
+
+ /**
+ * Send batch messages synchronously.
+ *
+ * <p>This method does not return until it gets the definitive result.
+ *
+ * <p>All messages to send should have the same topic.
+ *
+ * @param messages batch messages to send.
+ * @return a map which indicates the message id assigned to the appointed message.
+ * @throws TopicDoesNotExistException if the topic of message does not exist.
+ * @throws AuthorisationException if no permission to send message.
+ * @throws AuthenticationException if identification could not be recognized by server.
+ * @throws ProducerClosedAlreadyException if producer is closed already.
+ * @throws MessageTypeDoesNotMatchException if message type does not match with the topic.
+ * @throws NetworkTimeoutException if encountered network timeout to communicate with server.
+ * @throws NetworkConnectionException if there is a network connection problem.
+ * @throws PersistenceException if encountered persistence failure from server.
+ */
+ Map<Message, MessageId> send(Collection<Message> messages) throws ClientException;
Review comment:
What will happen if I use a list with the same object?
The return type does not match the argument.
Maybe both using list is ok.
##########
File path: apis/src/main/java/org/apache/rocketmq/apis/producer/Producer.java
##########
@@ -0,0 +1,132 @@
+/*
+ * 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.rocketmq.apis.producer;
+
+import com.google.common.util.concurrent.Service;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.io.Closeable;
+import java.util.Collection;
+import org.apache.rocketmq.apis.exception.AuthenticationException;
+import org.apache.rocketmq.apis.exception.AuthorisationException;
+import org.apache.rocketmq.apis.exception.ClientException;
+import org.apache.rocketmq.apis.exception.MessageTypeDoesNotMatchException;
+import org.apache.rocketmq.apis.exception.PersistenceException;
+import org.apache.rocketmq.apis.exception.ProducerClosedAlreadyException;
+import org.apache.rocketmq.apis.exception.NetworkConnectionException;
+import org.apache.rocketmq.apis.exception.NetworkTimeoutException;
+import org.apache.rocketmq.apis.exception.TopicDoesNotExistException;
+import org.apache.rocketmq.apis.exception.TransactionCheckerNotSetException;
+import org.apache.rocketmq.apis.message.Message;
+import org.apache.rocketmq.apis.message.MessageId;
+
+/**
+ * Producer is a thread-safe rocketmq client which is used to publish messages.
+ *
+ * <p>On account of network timeout or other reasons, rocketmq producer only promised the at-least-once semantics.
+ * For producer, at-least-once semantics means potentially attempts are made at sending it, messages may be
+ * duplicated but not lost.
+ */
+public interface Producer extends Closeable {
+ /**
+ * Sends a message synchronously.
+ *
+ * <p>This method does not return until it gets the definitive result.
+ *
+ * @param message message to send.
+ * @return the message id assigned to the appointed message.
+ * @throws TopicDoesNotExistException if the topic of message does not exist.
+ * @throws AuthorisationException if no permission to send message.
+ * @throws AuthenticationException if identification could not be recognized by server.
+ * @throws ProducerClosedAlreadyException if producer is closed already.
+ * @throws MessageTypeDoesNotMatchException if message type does not match with the topic.
+ * @throws NetworkTimeoutException if encountered network timeout to communicate with server.
+ * @throws NetworkConnectionException if there is a network connection problem.
+ * @throws PersistenceException if encountered persistence failure from server.
+ */
+ MessageId send(Message message) throws ClientException;
Review comment:
Where is the SendResult?
The client at least needs to know the offset and messagequeue for checking!
##########
File path: apis/src/main/java/org/apache/rocketmq/apis/message/Message.java
##########
@@ -0,0 +1,77 @@
+/*
+ * 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.rocketmq.apis.message;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.Optional;
+import org.apache.rocketmq.apis.producer.Producer;
+
+/**
+ * Abstract message only used for {@link Producer}.
+ */
+public interface Message {
+ /**
+ * Get the topic of message, which is the first classifier for message.
+ *
+ * @return topic of message.
+ */
+ String getTopic();
+
+ /**
+ * Get the <strong>deep copy</strong> of message body.
+ *
+ * @return the <strong>deep copy</strong> of message body.
+ */
+ byte[] getBody();
+
+ /**
+ * Get the <strong>deep copy</strong> of message properties.
+ *
+ * @return the <strong>deep copy</strong> of message properties.
+ */
+ Map<String, String> getProperties();
+
+ /**
+ * Get the tag of message, which is the second classifier besides topic.
+ *
+ * @return the tag of message.
+ */
+ Optional<String> getTag();
+
+ /**
+ * Get the key collection of message.
+ *
+ * @return <strong>the key collection</strong> of message.
+ */
+ Collection<String> getKeys();
+
+ /**
+ * Get the message group, which make sense only when topic type is fifo.
+ *
+ * @return message group, which is optional.
+ */
+ Optional<String> getMessageGroup();
+
Review comment:
This interface need a method getMessageQueue() too.
It may need to send message to the specified MessageQueue.
The case cannot be covered by getMessageGroup().
The priority of getMessageQueue() is bigger than getMessageGroup().
##########
File path: apis/src/main/java/org/apache/rocketmq/apis/producer/Producer.java
##########
@@ -0,0 +1,132 @@
+/*
+ * 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.rocketmq.apis.producer;
+
+import com.google.common.util.concurrent.Service;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.io.Closeable;
+import java.util.Collection;
+import org.apache.rocketmq.apis.exception.AuthenticationException;
+import org.apache.rocketmq.apis.exception.AuthorisationException;
+import org.apache.rocketmq.apis.exception.ClientException;
+import org.apache.rocketmq.apis.exception.MessageTypeDoesNotMatchException;
+import org.apache.rocketmq.apis.exception.PersistenceException;
+import org.apache.rocketmq.apis.exception.ProducerClosedAlreadyException;
+import org.apache.rocketmq.apis.exception.NetworkConnectionException;
+import org.apache.rocketmq.apis.exception.NetworkTimeoutException;
+import org.apache.rocketmq.apis.exception.TopicDoesNotExistException;
+import org.apache.rocketmq.apis.exception.TransactionCheckerNotSetException;
+import org.apache.rocketmq.apis.message.Message;
+import org.apache.rocketmq.apis.message.MessageId;
+
+/**
+ * Producer is a thread-safe rocketmq client which is used to publish messages.
+ *
+ * <p>On account of network timeout or other reasons, rocketmq producer only promised the at-least-once semantics.
+ * For producer, at-least-once semantics means potentially attempts are made at sending it, messages may be
+ * duplicated but not lost.
+ */
+public interface Producer extends Closeable {
+ /**
+ * Sends a message synchronously.
+ *
+ * <p>This method does not return until it gets the definitive result.
+ *
+ * @param message message to send.
+ * @return the message id assigned to the appointed message.
+ * @throws TopicDoesNotExistException if the topic of message does not exist.
+ * @throws AuthorisationException if no permission to send message.
+ * @throws AuthenticationException if identification could not be recognized by server.
+ * @throws ProducerClosedAlreadyException if producer is closed already.
+ * @throws MessageTypeDoesNotMatchException if message type does not match with the topic.
+ * @throws NetworkTimeoutException if encountered network timeout to communicate with server.
+ * @throws NetworkConnectionException if there is a network connection problem.
+ * @throws PersistenceException if encountered persistence failure from server.
+ */
+ MessageId send(Message message) throws ClientException;
+
+ /**
+ * Sends a transactional message synchronously.
+ *
+ * @param message message to send.
+ * @param transaction transaction to bind.
+ * @return the message id assigned to the appointed message.
+ * @throws TopicDoesNotExistException if the topic of message does not exist.
+ * @throws AuthorisationException if no permission to send message.
+ * @throws AuthenticationException if identification could not be recognized by server.
+ * @throws ProducerClosedAlreadyException if producer is closed already.
+ * @throws MessageTypeDoesNotMatchException if message type does not match with the topic.
+ * @throws NetworkTimeoutException if encountered network timeout to communicate with server.
+ * @throws NetworkConnectionException if there is a network connection problem.
+ * @throws PersistenceException if encountered persistence failure from server.
+ * @throws TransactionCheckerNotSetException if {@link TransactionChecker} is not set.
+ */
+ MessageId send(Message message, Transaction transaction) throws ClientException;
Review comment:
This is not the two-phase API!
How about like this :
try {
TransactionMark mark = producer.prepareSend(Message message);
//do process
producer.commit(mark);
} catch(Exception e) {
producer.rollback(mark);
}
Such style is easy to be integrated with db transaction. And the code will not be split.
--
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@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org