Address PR comments

test/ziggurat/messaging/consumer_test.clj

@@ -312,9 +312,10 @@
 
 (deftest convert-and-ack-message-test
   (let [message                    {:id 7 :path "/photos/h2k3j4h9h23"}
+        metadata                   {:topic "x" :partition 1 :timestamp 123}
         proto-class                Example$Photo
         proto-message              (proto/->bytes (proto/create proto-class message))
-        message-payload            {:message proto-message :topic-entity topic-entity :retry-count 3}]
+        message-payload            {:message proto-message :topic-entity topic-entity :retry-count 3 :metadata metadata}]
     (testing "should return deserialized message-payload if serialized using protobuf"
       (let [proto-serialized-message-payload (zmd/serialize-to-message-payload-proto message-payload)
             converted-message-payload        (consumer/convert-and-ack-message nil {:delivery-tag 1} proto-serialized-message-payload false "default")]
@@ -333,13 +334,13 @@
       (let [nippy-message-payload         (assoc (zmp/->MessagePayload proto-message topic-entity) :retry-count 3)
             nippy-serialized-message-payload (nippy/freeze nippy-message-payload)
             converted-message-payload        (consumer/convert-and-ack-message nil {:delivery-tag 1} nippy-serialized-message-payload false "default")]
-        (is (= (bytes-to-str converted-message-payload) (bytes-to-str message-payload)))))
+        (is (= (bytes-to-str converted-message-payload) (bytes-to-str (into {} nippy-message-payload))))))
     (testing "should return a map and not ziggurat.mapper/->MessagePayload if original message was a ziggurat.mapper/->MessagePayload and serialized using nippy"
       (let [nippy-message-payload         (assoc (mpr/->MessagePayload proto-message topic-entity) :retry-count 3)
             nippy-serialized-message-payload (nippy/freeze nippy-message-payload)
             converted-message-payload        (consumer/convert-and-ack-message nil {:delivery-tag 1} nippy-serialized-message-payload false "default")]
-        (is (= (bytes-to-str converted-message-payload) (bytes-to-str message-payload)))))
-    (testing "should return a clojure map if serialized using nippy"
+        (is (= (bytes-to-str converted-message-payload) (bytes-to-str (into {} nippy-message-payload))))))
+    (testing "should return nil if the message is serialized using neither proto or nippy"
       (let [random-bytes-as-message-payload     (.getBytes (String. "Hello World"))
             converted-message-payload           (consumer/convert-and-ack-message nil {:delivery-tag 1} random-bytes-as-message-payload false "default")]
         (is (= converted-message-payload nil))))))

test/ziggurat/messaging/dead_set_test.clj

@@ -11,7 +11,10 @@
                                     fix/mount-metrics]))
 
 (def topic-entity :default)
-(def default-message-payload {:message (.getBytes "foo-bar") :topic-entity topic-entity :retry-count 1})
+(def default-message-payload {:message (.getBytes "foo-bar")
+                              :topic-entity topic-entity
+                              :retry-count 1
+                              :metadata {:topic "x" :partition 1 :timestamp 123}})
 
 (deftest replay-test
   (testing "puts message from dead set to instant queue"

test/ziggurat/messaging/producer_test.clj

@@ -23,7 +23,7 @@
                                     fix/silence-logging]))
 
 (def topic-entity :default)
-(def message-payload {:message (.getBytes "hello-world") :topic-entity topic-entity})
+(def message-payload {:message (.getBytes "hello-world") :topic-entity topic-entity :metadata {:topic "x" :partition 1 :timestamp 123}})
 (defn retry-count-config [] (-> (ziggurat-config) :retry :count))
 
 (deftest retry-for-channel-test
@@ -158,6 +158,8 @@
         (let [message-from-mq (rmq/get-msg-from-dead-queue "default")]
           (is (= nil message-from-mq)))
         (let [message-from-mq (rmq/get-msg-from-delay-queue "default")]
+          (is (= nil message-from-mq)))
+        (let [message-from-mq (rmq/get-msg-from-instant-queue "default")]
           (is (= nil message-from-mq))))))
 
   (testing "publish to delay queue publishes with expiration from config"
@@ -220,6 +222,17 @@
                                                      :retry {:count   5
                                                              :enabled true
                                                              :type    :exponential}))]
+
+      (testing "message with available retry counts same as that of ziggurat-config will be published to delay queue with suffix 1"
+        (fix/with-queues
+          {:default {:handler-fn #(constantly nil)}}
+          (let [retry-message-payload (assoc message-payload :retry-count (get-in (ziggurat-config) [:retry :count]))
+                expected-message      (assoc message-payload :retry-count 4)]
+            (producer/retry retry-message-payload)
+            (let [message-from-mq (rmq/get-message-from-retry-queue "default" 1)]
+              (is (= (bytes-to-str message-from-mq) (bytes-to-str expected-message)))))))
+
+
       (testing "message with available retry counts as 4 will be published to delay queue with suffix 2"
         (fix/with-queues
           {:default {:handler-fn #(constantly nil)}}
@@ -612,7 +625,7 @@
     (let [serialize-called (atom false)]
       (with-redefs [zmd/serialize-to-message-payload-proto (fn [_] (reset! serialize-called true))
                     lch/open                               (fn [^Connection _] (reify Channel (close [_] nil)))
-                    lb/publish                             (fn [_ _ _ _ _])]
+                    lb/publish                             (fn [_ _ _ _ _] nil)]
         (producer/publish "exchange" message-payload)
         (is (true? @serialize-called))))))