Handle TopicTerminated for producers to fail immediately rather than retrying and timing out.

Author: planetbeing

lib/HandlerBase.cc

@@ -171,6 +171,7 @@ void HandlerBase::handleDisconnection(Result result, const ClientConnectionPtr&
         case Closing:
         case Closed:
         case Producer_Fenced:
+        case Terminated:
         case Failed:
             LOG_DEBUG(getName() << "Ignoring connection closed event since the handler is not used anymore");
             break;

lib/HandlerBase.h

@@ -138,6 +138,8 @@ class HandlerBase : public std::enable_shared_from_this<HandlerBase> {
         Failed,           // Handler is failed, in Java client: HandlerState.State.Failed
         Producer_Fenced,  // The producer has been fenced by the broker
                           // in Java client: HandlerState.State.ProducerFenced
+        Terminated,       // The topic has been terminatedproducer has been fenced by the broker
+                          // in Java client: HandlerState.State.Terminated
     };
 
     std::atomic<State> state_;

lib/ProducerImpl.cc

@@ -273,8 +273,8 @@ Result ProducerImpl::handleCreateProducer(const ClientConnectionPtr& cnx, Result
             }
         }
 
-        if (result == ResultProducerFenced) {
-            state_ = Producer_Fenced;
+        if (result == ResultProducerFenced || result == ResultTopicTerminated) {
+            state_ = result == ResultProducerFenced ? Producer_Fenced : Terminated;
             failPendingMessages(result, false);
             auto client = client_.lock();
             if (client) {
@@ -450,6 +450,9 @@ bool ProducerImpl::isValidProducerState(const SendCallback& callback) const {
         case HandlerBase::Producer_Fenced:
             callback(ResultProducerFenced, {});
             return false;
+        case HandlerBase::Terminated:
+            callback(ResultTopicTerminated, {});
+            return false;
         case HandlerBase::NotStarted:
         case HandlerBase::Failed:
         default:

lib/ResultUtils.h

@@ -39,6 +39,7 @@ inline bool isResultRetryable(Result result) {
                                                       ResultInvalidConfiguration,
                                                       ResultIncompatibleSchema,
                                                       ResultTopicNotFound,
+                                                      ResultTopicTerminated,
                                                       ResultOperationNotSupported,
                                                       ResultNotAllowedError,
                                                       ResultChecksumError,