Force push_mode to be passed as a keyword argument for input adapters (#657)

* Fix nasty bug where Kafka adapter in Burst mode gave wrong Edge type

Signed-off-by: Adam Glustein <adam.glustein@point72.com>

* Force push_mode to be passed as a keyword argument, error if it is passed positionally

Signed-off-by: Adam Glustein <adam.glustein@point72.com>

---------

Signed-off-by: Adam Glustein <adam.glustein@point72.com>

Author: AdamGlustein

csp/adapters/kafka.py

@@ -187,7 +187,7 @@ def subscribe(
                 )
             properties["tick_timestamp_from_field"] = tick_timestamp_from_field
 
-        return _kafka_input_adapter_def(self, ts_type, properties, push_mode)
+        return _kafka_input_adapter_def(self, ts_type, properties, push_mode=push_mode)
 
     def publish(
         self,
@@ -225,7 +225,7 @@ def publish(
 
     def status(self, push_mode=csp.PushMode.NON_COLLAPSING):
         ts_type = Status
-        return status_adapter_def(self, ts_type, push_mode)
+        return status_adapter_def(self, ts_type, push_mode=push_mode)
 
     def __hash__(self):
         return hash((self._group_id_prefix, hash_mutable(self._properties)))

csp/adapters/parquet.py

@@ -268,7 +268,7 @@ def subscribe_dict_basket_struct_column(
 
     def status(self, push_mode=PushMode.NON_COLLAPSING):
         ts_type = Status
-        return status_adapter_def(self, ts_type, push_mode)
+        return status_adapter_def(self, ts_type, push_mode=push_mode)
 
     def _create(self, engine, memo):
         """method needs to return the wrapped c++ adapter manager"""

csp/impl/wiring/adapters.py

@@ -43,7 +43,9 @@ def _instantiate_impl(cls, __forced_tvars, name, args, kwargs):
     def _instantiate(self, __forced_tvars, name, *args, **kwargs):
         return self._instantiate_func(__forced_tvars, name, args=args, kwargs=kwargs)
 
-    def __call__(cls, *args, **kwargs):
+    def __call__(cls, *args, push_mode=PushMode.NON_COLLAPSING, **kwargs):
+        if cls.is_input:
+            kwargs["push_mode"] = push_mode
         return cls._instantiate(None, None, *args, **kwargs)
 
     def using(cls, name=None, **__forced_tvars):
@@ -183,6 +185,7 @@ def _adapterdef(BaseDef, name, create_method, out_type, manager_type, memoize=Tr
             "memoize": memoize,
             "force_memoize": force_memoize,
             "_impl": create_method,
+            "is_input": is_input,
         },
     )
     return adaptertype

csp/tests/adapters/test_kafka.py

@@ -1,4 +1,5 @@
 import os
+import typing
 from datetime import date, datetime, timedelta
 
 import pytest
@@ -514,6 +515,11 @@ def sub_graph():
                     key="foo",
                     push_mode=push_mode,
                 )
+                # ensure graph-time ts type is correct
+                if push_mode == csp.PushMode.BURST:
+                    assert data.tstype.typ is typing.List[BasicData]
+                else:
+                    assert data.tstype.typ is BasicData
                 csp.add_graph_output(key, data)
                 stop_flags.append(csp.count(data) == 1)