Fix Arbitrary Properties of kombu Producer.publish (#1544)

* Fix arbitrary properties of kombu Producer.publish

* Add regression test

Author: TimPansino

newrelic/hooks/messagebroker_kombu.py

@@ -82,7 +82,7 @@ def bind_publish(
         "expiration": expiration,
         "timeout": timeout,
         "confirm_timeout": confirm_timeout,
-        "properties": properties,
+        **properties,
     }
 
 

tests/datastore_rediscluster/test_uninstrumented_rediscluster_methods.py

@@ -138,6 +138,7 @@
     "result_callbacks",
     "set_default_node",
     "user_on_connect_func",
+    "startup_nodes",
 }
 
 REDIS_MODULES = {"bf", "cf", "cms", "ft", "graph", "json", "tdigest", "topk", "ts", "vset"}

tests/messagebroker_kombu/conftest.py

@@ -12,8 +12,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import json
-import pickle
 import uuid
 
 import kombu
@@ -109,7 +107,7 @@ def consumer_validate_dt(producer, consumer_connection, queue, consume_validate_
 def consume(events):
     def _consume(body, message):
         message.ack()
-        events.append({"body": body, "routing_key": message.delivery_info["routing_key"]})
+        events.append({"body": body, "routing_key": message.delivery_info["routing_key"], "message": message})
 
     return _consume
 
@@ -118,7 +116,7 @@ def _consume(body, message):
 def consume_error(events):
     def _consume(body, message):
         message.ack()
-        events.append({"body": body, "routing_key": message.delivery_info["routing_key"]})
+        events.append({"body": body, "routing_key": message.delivery_info["routing_key"], "message": message})
         raise RuntimeError("Error in consumer callback")
 
     return _consume
@@ -135,7 +133,7 @@ def _consume(body, message):
         txn._test_request_headers = message.headers
 
         message.ack()
-        events.append({"body": body, "routing_key": message.delivery_info["routing_key"]})
+        events.append({"body": body, "routing_key": message.delivery_info["routing_key"], "message": message})
 
     return _consume
 

tests/messagebroker_kombu/test_producer.py

@@ -22,6 +22,7 @@
 from newrelic.api.background_task import background_task
 from newrelic.api.function_trace import FunctionTrace
 from newrelic.common.object_names import callable_name
+from newrelic.common.object_wrapper import transient_function_wrapper
 
 
 def test_trace_metrics(send_producer_message, exchange):
@@ -105,3 +106,24 @@ def test():
             )
 
     test()
+
+
+def test_custom_properties(exchange, queue, events, get_consumer_record, producer):
+    validate_custom_properties_called = []
+
+    @transient_function_wrapper("kombu.messaging", "Producer._publish")
+    def validate_custom_properties(wrapped, instance, args, kwargs):
+        from newrelic.common.signature import bind_args
+
+        bound_args = bind_args(wrapped, args, kwargs)
+        properties = bound_args["properties"]
+        assert properties.get("custom_property", "") == "baz", "custom_property was deleted by instrumentation."
+        validate_custom_properties_called.append(True)
+
+    @background_task()
+    @validate_custom_properties
+    def test():
+        producer.publish({"foo": 123}, exchange=exchange, routing_key="bar", declare=[queue], custom_property="baz")
+
+    test()
+    assert validate_custom_properties_called, "validate_custom_properties was not called."

tox.ini

@@ -68,7 +68,8 @@ envlist =
     {windows,windows_arm64}-cross_agent-{py313,py314}-{with,without}_extensions,
 
     # Integration Tests (only run on Linux)
-    cassandra-datastore_cassandradriver-{py38,py39,py310,py311,py312,pypy311}-cassandralatest,
+    cassandra-datastore_cassandradriver-py38-cassandra032903,
+    cassandra-datastore_cassandradriver-{py39,py310,py311,py312,pypy311}-cassandralatest,
     elasticsearchserver07-datastore_elasticsearch-{py38,py39,py310,py311,py312,py313,py314,pypy311}-elasticsearch07,
     elasticsearchserver08-datastore_elasticsearch-{py38,py39,py310,py311,py312,py313,py314,pypy311}-elasticsearch08,
     firestore-datastore_firestore-{py38,py39,py310,py311,py312,py313,py314},
@@ -283,7 +284,8 @@ deps =
     datastore_aiomysql: sqlalchemy<2
     datastore_bmemcached: python-binary-memcached
     datastore_cassandradriver-cassandralatest: cassandra-driver
-    datastore_cassandradriver-cassandralatest: twisted
+    datastore_cassandradriver-cassandra032903: cassandra-driver<3.29.3
+    datastore_cassandradriver: twisted
     datastore_elasticsearch: requests
     datastore_elasticsearch: httpx
     datastore_elasticsearch-elasticsearch07: elasticsearch[async]<8.0