Enable Kafka transport for operator mode (#49)

* Add kafka transport for operator mode.

* Skip health check for Kafka transport.

* Fix kafka transport example.

* Do not perform readiness check for kafka transport.

Author: doru1004

examples/kafka/kafka_event_streaming_source.py

@@ -61,7 +61,7 @@
     kind='http-source',
     url='https://query1.finance.yahoo.com/v7/finance/quote?symbols=AAPL',
     route='/from-http',
-    period=3000,
+    period=1000,
     kafka_transport_topic=topic,
     kafka_transport_partitions=3)
 

rayvens/api.py

@@ -21,6 +21,7 @@
 from rayvens.core.local import start as start_http
 from rayvens.core.kafka import start as start_kafka
 from rayvens.core.operator import start as start_operator_http
+from rayvens.core.operator_kafka import start as start_operator_kafka
 from rayvens.core.ray_serve import start as start_operator_ray_serve
 from rayvens.core.name import name_source, name_sink
 from rayvens.core.verify import verify_do
@@ -252,6 +253,8 @@ def init(mode=os.getenv('RAYVENS_MODE', 'auto'),
     elif mode in ['mixed', 'operator']:
         if transport in ['auto', 'http']:
             _global_camel = start_operator_http(mode, check_port)
+        elif transport == 'kafka':
+            _global_camel = start_operator_kafka(mode, check_port)
         elif transport in ['ray-serve']:
             _global_camel = start_operator_ray_serve(mode, check_port)
         else:

rayvens/core/common.py

@@ -96,6 +96,10 @@ def await_start(mode, integration):
     else:
         print('Integration did not start correctly.')
 
+    # For kafka transport the health check cannot be performed.
+    if mode.transport == 'kafka':
+        return True
+
     # Perform health check and wait for integration to be ready.
     healthy_integration = _wait_for_ready_integration(mode, integration)
 
@@ -276,6 +280,12 @@ def brokers():
     return f'{host}:{port}'
 
 
+def operator_brokers():
+    host = os.getenv('KAFKA_SERVICE_HOST', 'kafka')
+    port = os.getenv('KAFKA_SERVICE_PORT', '9092')
+    return f'{host}:{port}'
+
+
 def eval(f, data):
     if isinstance(f, ActorHandle):
         return f.append.remote(data)

rayvens/core/operator_kafka.py

@@ -0,0 +1,82 @@
+#
+# Copyright IBM Corporation 2021
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+from rayvens.core.common import get_run_mode, await_start, brokers
+from rayvens.core.common import kafka_send_to, kafka_recv_from
+from rayvens.core.catalog import construct_source, construct_sink
+from rayvens.core.integration import Integration
+
+
+def start(camel_mode, check_port):
+    return Camel(get_run_mode(camel_mode, check_port))
+
+
+class Camel:
+    def __init__(self, mode):
+        self.mode = mode
+        self.mode.transport = 'kafka'
+
+    def add_source(self, stream, source, source_name):
+        # Construct integration
+        integration = Integration(stream.name, source_name, source)
+
+        # Prepare env:
+        integration.prepare_environment(self.mode)
+
+        # Determine the `to` endpoint value made up of a base address and
+        # a custom route provided by the user. Use this to construct the
+        # integration source code.
+        integration_content = construct_source(
+            source,
+            f'kafka:{integration.kafka_transport_topic}?brokers={brokers()}')
+
+        # Start running the source integration.
+        integration.invoke_run(self.mode, integration_content)
+
+        # Set up source for the HTTP connector case.
+        kafka_send_to(integration.kafka_transport_topic,
+                      integration.kafka_transport_partitions, stream.actor)
+
+        if not await_start(self.mode, integration):
+            raise RuntimeError('Could not start source')
+        return integration
+
+    def add_sink(self, stream, sink, sink_name):
+        # Construct integration
+        integration = Integration(stream.name, sink_name, sink)
+
+        # Prepare env:
+        integration.prepare_environment(self.mode)
+
+        # Get integration source code.
+        integration_content = construct_sink(
+            sink,
+            f'kafka:{integration.kafka_transport_topic}?brokers={brokers()}')
+
+        # Start running the integration.
+        integration.invoke_run(self.mode, integration_content)
+
+        kafka_recv_from(sink_name, integration.kafka_transport_topic,
+                        stream.actor)
+
+        # Wait for integration to finish.
+        if not await_start(self.mode, integration):
+            raise RuntimeError('Could not start sink')
+
+        return integration
+
+    def disconnect(self, integration):
+        integration.disconnect(self.mode)