Skip to content

feat: add a flag to retry another broker when not store ok #457

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 2 commits into
base: master
Choose a base branch
from
+59 −5
Open

feat: add a flag to retry another broker when not store ok #457

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f0f65a6
feat: add a flag to retry another broker when not store ok
ifplusor Oct 6, 2023
bc63a8f
feat: expose c api
ifplusor Oct 6, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .gitignore
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
.idea
.cache/
cmake-build-debug/
bin
build
Expand Down
3 changes: 2 additions & 1 deletion include/CProducer.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,7 @@ ROCKETMQCLIENT_API int SetProducerSendMsgTimeout(CProducer* producer, int timeou
ROCKETMQCLIENT_API int SetProducerCompressLevel(CProducer* producer, int level);
ROCKETMQCLIENT_API int SetProducerMaxMessageSize(CProducer* producer, int size);
ROCKETMQCLIENT_API int SetProducerMessageTrace(CProducer* consumer, CTraceModel openTrace);
ROCKETMQCLIENT_API int SetProducerRetryAnotherBrokerWhenNotStoreOK(CProducer* producer, int retry);

ROCKETMQCLIENT_API int SendMessageSync(CProducer* producer, CMessage* msg, CSendResult* result);
ROCKETMQCLIENT_API int SendBatchMessage(CProducer* producer, CBatchMessage* msg, CSendResult* result);
Expand Down Expand Up @@ -106,4 +107,4 @@ ROCKETMQCLIENT_API int SendMessageTransaction(CProducer* producer,
#ifdef __cplusplus
}
#endif
#endif //__C_PRODUCER_H__
#endif //__C_PRODUCER_H__
3 changes: 3 additions & 0 deletions include/DefaultMQProducer.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -98,6 +98,9 @@ class ROCKETMQCLIENT_API DefaultMQProducer {
int getMaxMessageSize() const;
void setMaxMessageSize(int maxMessageSize);

bool getRetryAnotherBrokerWhenNotStoreOK() const;
void setRetryAnotherBrokerWhenNotStoreOK(bool retry);

int getRetryTimes() const;
void setRetryTimes(int times);

Expand Down
30 changes: 30 additions & 0 deletions src/extern/CProducer.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -218,6 +218,7 @@ typedef struct __DefaultProducer__ {
int producerType;
char* version;
} DefaultProducer;

CProducer* CreateProducer(const char* groupId) {
if (groupId == NULL) {
return NULL;
Expand Down Expand Up @@ -267,6 +268,7 @@ CProducer* CreateTransactionProducer(const char* groupId, CLocalTransactionCheck
defaultMQProducer->version[MAX_SDK_VERSION_LENGTH - 1] = 0;
return (CProducer*)defaultMQProducer;
}

int DestroyProducer(CProducer* pProducer) {
if (pProducer == NULL) {
return NULL_POINTER;
Expand All @@ -290,6 +292,7 @@ int DestroyProducer(CProducer* pProducer) {
delete reinterpret_cast<DefaultProducer*>(pProducer);
return OK;
}

int StartProducer(CProducer* producer) {
if (producer == NULL) {
return NULL_POINTER;
Expand All @@ -307,6 +310,7 @@ int StartProducer(CProducer* producer) {
}
return OK;
}

int ShutdownProducer(CProducer* producer) {
if (producer == NULL) {
return NULL_POINTER;
Expand All @@ -324,6 +328,7 @@ int ShutdownProducer(CProducer* producer) {
}
return OK;
}

const char* ShowProducerVersion(CProducer* producer) {
if (producer == NULL) {
return DEFAULT_SDK_VERSION;
Expand All @@ -332,6 +337,7 @@ const char* ShowProducerVersion(CProducer* producer) {

return defaultMQProducer->version;
}

int SetProducerNameServerAddress(CProducer* producer, const char* namesrv) {
if (producer == NULL) {
return NULL_POINTER;
Expand Down Expand Up @@ -577,6 +583,7 @@ int SendMessageOrderly(CProducer* producer,
}
return OK;
}

int SendMessageOrderlyByShardingKey(CProducer* producer, CMessage* msg, const char* shardingKey, CSendResult* result) {
if (producer == NULL || msg == NULL || shardingKey == NULL || result == NULL) {
return NULL_POINTER;
Expand Down Expand Up @@ -629,6 +636,7 @@ int SendMessageTransaction(CProducer* producer,
}
return OK;
}

int SetProducerGroupName(CProducer* producer, const char* groupName) {
if (producer == NULL) {
return NULL_POINTER;
Expand Down Expand Up @@ -663,6 +671,7 @@ int SetProducerInstanceName(CProducer* producer, const char* instanceName) {
}
return OK;
}

int SetProducerSessionCredentials(CProducer* producer,
const char* accessKey,
const char* secretKey,
Expand All @@ -683,6 +692,7 @@ int SetProducerSessionCredentials(CProducer* producer,
}
return OK;
}

int SetProducerLogPath(CProducer* producer, const char* logPath) {
if (producer == NULL) {
return NULL_POINTER;
Expand Down Expand Up @@ -798,6 +808,7 @@ int SetProducerMaxMessageSize(CProducer* producer, int size) {
}
return OK;
}

int SetProducerMessageTrace(CProducer* producer, CTraceModel openTrace) {
if (producer == NULL) {
return NULL_POINTER;
Expand All @@ -816,6 +827,25 @@ int SetProducerMessageTrace(CProducer* producer, CTraceModel openTrace) {
}
return OK;
}

int SetProducerRetryAnotherBrokerWhenNotStoreOK(CProducer* producer, int retry) {
if (producer == NULL) {
return NULL_POINTER;
}

DefaultProducer* defaultMQProducer = (DefaultProducer*)producer;

try {
if (CAPI_C_PRODUCER_TYPE_TRANSACTION != defaultMQProducer->producerType) {
defaultMQProducer->innerProducer->setRetryAnotherBrokerWhenNotStoreOK(retry != 0);
}
} catch (exception& e) {
MQClientErrorContainer::setErr(string(e.what()));
return PRODUCER_START_FAILED;
}
return OK;
}

#ifdef __cplusplus
};
#endif
10 changes: 9 additions & 1 deletion src/producer/DefaultMQProducer.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -135,6 +135,14 @@ int DefaultMQProducer::getMaxMessageSize() const {
return impl->getMaxMessageSize();
}

bool DefaultMQProducer::getRetryAnotherBrokerWhenNotStoreOK() const {
return impl->getRetryAnotherBrokerWhenNotStoreOK();
}

void DefaultMQProducer::setRetryAnotherBrokerWhenNotStoreOK(bool retry) {
impl->setRetryAnotherBrokerWhenNotStoreOK(retry);
}

void DefaultMQProducer::setRetryTimes4Async(int times) {
impl->setRetryTimes4Async(times);
}
Expand Down Expand Up @@ -254,4 +262,4 @@ const std::string& DefaultMQProducer::getSslPropertyFile() const {
return impl->getSslPropertyFile();
}

} // namespace rocketmq
} // namespace rocketmq
12 changes: 10 additions & 2 deletions src/producer/DefaultMQProducerImpl.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,7 @@ DefaultMQProducerImpl::DefaultMQProducerImpl(const string& groupname)
: m_sendMsgTimeout(3000),
m_compressMsgBodyOverHowmuch(4 * 1024),
m_maxMessageSize(1024 * 128),
// m_retryAnotherBrokerWhenNotStoreOK(false),
m_retryAnotherBrokerWhenNotStoreOK(false),
m_compressLevel(5),
m_retryTimes(5),
m_retryTimes4Async(1),
Expand Down Expand Up @@ -410,7 +410,7 @@ SendResult DefaultMQProducerImpl::sendDefaultImpl(MQMessage& msg,
case ComMode_ONEWAY:
return sendResult;
case ComMode_SYNC:
if (sendResult.getSendStatus() != SEND_OK) {
if (sendResult.getSendStatus() != SEND_OK && m_retryAnotherBrokerWhenNotStoreOk) {
Copy link
Member

@yuz10 yuz10 Oct 31, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The default behavior is changed, because default m_retryAnotherBrokerWhenNotStoreOk is false, should add docs

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe. In Java it is false, and I keep it consistent.

https://github.com/apache/rocketmq/blob/26fa0501482bbf31c2a64a33f329ab9744ac3800/client/src/main/java/org/apache/rocketmq/client/producer/DefaultMQProducer.java#L124-L127

if (bActiveMQ) {
topicPublishInfo->updateNonServiceMessageQueue(mq, getSendMsgTimeout());
}
Expand Down Expand Up @@ -620,6 +620,14 @@ bool DefaultMQProducerImpl::tryToCompressMessage(MQMessage& msg) {
return false;
}

bool DefaultMQProducerImpl::getRetryAnotherBrokerWhenNotStoreOK() const {
return m_retryAnotherBrokerWhenNotStoreOK;
}

void DefaultMQProducerImpl::setRetryAnotherBrokerWhenNotStoreOK(bool retry) {
m_retryAnotherBrokerWhenNotStoreOK = retry;
}

int DefaultMQProducerImpl::getRetryTimes() const {
return m_retryTimes;
}
Expand Down
5 changes: 4 additions & 1 deletion src/producer/DefaultMQProducerImpl.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,6 +84,9 @@ class DefaultMQProducerImpl : public MQProducer {
void setRetryTimes4Async(int times);
void submitSendTraceRequest(const MQMessage& msg, SendCallback* pSendCallback);

bool getRetryAnotherBrokerWhenNotStoreOK() const;
void setRetryAnotherBrokerWhenNotStoreOK(bool retry);

protected:
SendResult sendAutoRetrySelectImpl(MQMessage& msg,
MessageQueueSelector* pSelector,
Expand Down Expand Up @@ -122,7 +125,7 @@ class DefaultMQProducerImpl : public MQProducer {
int m_sendMsgTimeout;
int m_compressMsgBodyOverHowmuch;
int m_maxMessageSize; //<! default:128K;
// bool m_retryAnotherBrokerWhenNotStoreOK;
bool m_retryAnotherBrokerWhenNotStoreOK;
int m_compressLevel;
int m_retryTimes;
int m_retryTimes4Async;
Expand Down