[jira] [Work logged] (BEAM-10125) adding cross-language KafkaIO integration test
[ https://issues.apache.org/jira/browse/BEAM-10125?focusedWorklogId=438893=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-438893 ] ASF GitHub Bot logged work on BEAM-10125: - Author: ASF GitHub Bot Created on: 29/May/20 18:34 Start Date: 29/May/20 18:34 Worklog Time Spent: 10m Work Description: chamikaramj merged pull request #11847: URL: https://github.com/apache/beam/pull/11847 This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 438893) Time Spent: 2.5h (was: 2h 20m) > adding cross-language KafkaIO integration test > -- > > Key: BEAM-10125 > URL: https://issues.apache.org/jira/browse/BEAM-10125 > Project: Beam > Issue Type: Improvement > Components: cross-language, io-java-kafka >Reporter: Heejong Lee >Assignee: Heejong Lee >Priority: P2 > Time Spent: 2.5h > Remaining Estimate: 0h > > adding cross-language KafkaIO integration test -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-10125) adding cross-language KafkaIO integration test
[ https://issues.apache.org/jira/browse/BEAM-10125?focusedWorklogId=438564=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-438564 ] ASF GitHub Bot logged work on BEAM-10125: - Author: ASF GitHub Bot Created on: 29/May/20 04:32 Start Date: 29/May/20 04:32 Worklog Time Spent: 10m Work Description: chamikaramj commented on pull request #11847: URL: https://github.com/apache/beam/pull/11847#issuecomment-635752517 Retest this please This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 438564) Time Spent: 2h 20m (was: 2h 10m) > adding cross-language KafkaIO integration test > -- > > Key: BEAM-10125 > URL: https://issues.apache.org/jira/browse/BEAM-10125 > Project: Beam > Issue Type: Improvement > Components: io-java-kafka >Reporter: Heejong Lee >Assignee: Heejong Lee >Priority: P2 > Time Spent: 2h 20m > Remaining Estimate: 0h > > adding cross-language KafkaIO integration test -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-10125) adding cross-language KafkaIO integration test
[ https://issues.apache.org/jira/browse/BEAM-10125?focusedWorklogId=438560=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-438560 ] ASF GitHub Bot logged work on BEAM-10125: - Author: ASF GitHub Bot Created on: 29/May/20 04:31 Start Date: 29/May/20 04:31 Worklog Time Spent: 10m Work Description: chamikaramj commented on pull request #11847: URL: https://github.com/apache/beam/pull/11847#issuecomment-635752138 Retest this please This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 438560) Time Spent: 2h (was: 1h 50m) > adding cross-language KafkaIO integration test > -- > > Key: BEAM-10125 > URL: https://issues.apache.org/jira/browse/BEAM-10125 > Project: Beam > Issue Type: Improvement > Components: io-java-kafka >Reporter: Heejong Lee >Assignee: Heejong Lee >Priority: P2 > Time Spent: 2h > Remaining Estimate: 0h > > adding cross-language KafkaIO integration test -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-10125) adding cross-language KafkaIO integration test
[ https://issues.apache.org/jira/browse/BEAM-10125?focusedWorklogId=438561=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-438561 ] ASF GitHub Bot logged work on BEAM-10125: - Author: ASF GitHub Bot Created on: 29/May/20 04:31 Start Date: 29/May/20 04:31 Worklog Time Spent: 10m Work Description: chamikaramj commented on pull request #11847: URL: https://github.com/apache/beam/pull/11847#issuecomment-635752197 Retest this please This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 438561) Time Spent: 2h 10m (was: 2h) > adding cross-language KafkaIO integration test > -- > > Key: BEAM-10125 > URL: https://issues.apache.org/jira/browse/BEAM-10125 > Project: Beam > Issue Type: Improvement > Components: io-java-kafka >Reporter: Heejong Lee >Assignee: Heejong Lee >Priority: P2 > Time Spent: 2h 10m > Remaining Estimate: 0h > > adding cross-language KafkaIO integration test -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-10125) adding cross-language KafkaIO integration test
[ https://issues.apache.org/jira/browse/BEAM-10125?focusedWorklogId=438532=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-438532 ] ASF GitHub Bot logged work on BEAM-10125: - Author: ASF GitHub Bot Created on: 29/May/20 00:58 Start Date: 29/May/20 00:58 Worklog Time Spent: 10m Work Description: chamikaramj commented on pull request #11847: URL: https://github.com/apache/beam/pull/11847#issuecomment-635695687 Retest this please This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 438532) Time Spent: 1h 50m (was: 1h 40m) > adding cross-language KafkaIO integration test > -- > > Key: BEAM-10125 > URL: https://issues.apache.org/jira/browse/BEAM-10125 > Project: Beam > Issue Type: Improvement > Components: io-java-kafka >Reporter: Heejong Lee >Assignee: Heejong Lee >Priority: P2 > Time Spent: 1h 50m > Remaining Estimate: 0h > > adding cross-language KafkaIO integration test -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-10125) adding cross-language KafkaIO integration test
[ https://issues.apache.org/jira/browse/BEAM-10125?focusedWorklogId=438523=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-438523 ] ASF GitHub Bot logged work on BEAM-10125: - Author: ASF GitHub Bot Created on: 28/May/20 23:51 Start Date: 28/May/20 23:51 Worklog Time Spent: 10m Work Description: chamikaramj commented on pull request #11847: URL: https://github.com/apache/beam/pull/11847#issuecomment-635676513 Retest this please This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 438523) Time Spent: 1h 40m (was: 1.5h) > adding cross-language KafkaIO integration test > -- > > Key: BEAM-10125 > URL: https://issues.apache.org/jira/browse/BEAM-10125 > Project: Beam > Issue Type: Improvement > Components: io-java-kafka >Reporter: Heejong Lee >Assignee: Heejong Lee >Priority: P2 > Time Spent: 1h 40m > Remaining Estimate: 0h > > adding cross-language KafkaIO integration test -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-10125) adding cross-language KafkaIO integration test
[ https://issues.apache.org/jira/browse/BEAM-10125?focusedWorklogId=438522=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-438522 ] ASF GitHub Bot logged work on BEAM-10125: - Author: ASF GitHub Bot Created on: 28/May/20 23:51 Start Date: 28/May/20 23:51 Worklog Time Spent: 10m Work Description: chamikaramj commented on pull request #11847: URL: https://github.com/apache/beam/pull/11847#issuecomment-635676452 Retest this please This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 438522) Time Spent: 1.5h (was: 1h 20m) > adding cross-language KafkaIO integration test > -- > > Key: BEAM-10125 > URL: https://issues.apache.org/jira/browse/BEAM-10125 > Project: Beam > Issue Type: Improvement > Components: io-java-kafka >Reporter: Heejong Lee >Assignee: Heejong Lee >Priority: P2 > Time Spent: 1.5h > Remaining Estimate: 0h > > adding cross-language KafkaIO integration test -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-10125) adding cross-language KafkaIO integration test
[ https://issues.apache.org/jira/browse/BEAM-10125?focusedWorklogId=438505=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-438505 ] ASF GitHub Bot logged work on BEAM-10125: - Author: ASF GitHub Bot Created on: 28/May/20 23:30 Start Date: 28/May/20 23:30 Worklog Time Spent: 10m Work Description: chamikaramj commented on pull request #11847: URL: https://github.com/apache/beam/pull/11847#issuecomment-635670461 LGTM. Thanks. This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 438505) Time Spent: 1h 20m (was: 1h 10m) > adding cross-language KafkaIO integration test > -- > > Key: BEAM-10125 > URL: https://issues.apache.org/jira/browse/BEAM-10125 > Project: Beam > Issue Type: Improvement > Components: io-java-kafka >Reporter: Heejong Lee >Assignee: Heejong Lee >Priority: P2 > Time Spent: 1h 20m > Remaining Estimate: 0h > > adding cross-language KafkaIO integration test -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-10125) adding cross-language KafkaIO integration test
[
https://issues.apache.org/jira/browse/BEAM-10125?focusedWorklogId=438468=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-438468
]
ASF GitHub Bot logged work on BEAM-10125:
-
Author: ASF GitHub Bot
Created on: 28/May/20 21:47
Start Date: 28/May/20 21:47
Worklog Time Spent: 10m
Work Description: ihji commented on a change in pull request #11847:
URL: https://github.com/apache/beam/pull/11847#discussion_r432142992
##
File path: sdks/python/apache_beam/io/external/xlang_kafkaio_it_test.py
##
@@ -0,0 +1,145 @@
+"""Integration test for Python cross-language pipelines for Java KafkaIO."""
+
+from __future__ import absolute_import
+
+import contextlib
+import logging
+import os
+import socket
+import subprocess
+import time
+import typing
+import unittest
+
+import grpc
+
+import apache_beam as beam
+from apache_beam.io.external.kafka import ReadFromKafka
+from apache_beam.io.external.kafka import WriteToKafka
+from apache_beam.metrics import Metrics
+from apache_beam.options.pipeline_options import PipelineOptions
+from apache_beam.testing.test_pipeline import TestPipeline
+
+
+class CrossLanguageKafkaIO(object):
+ def __init__(self, bootstrap_servers, topic, expansion_service=None):
+self.bootstrap_servers = bootstrap_servers
+self.topic = topic
+self.expansion_service = expansion_service or (
+'localhost:%s' % os.environ.get('EXPANSION_PORT'))
+self.sum_counter = Metrics.counter('source', 'elements_sum')
+
+ def build_write_pipeline(self, pipeline):
+_ = (
+pipeline
+| 'Impulse' >> beam.Impulse()
+| 'Generate' >> beam.FlatMap(lambda x: range(1000)) # pylint:
disable=range-builtin-not-iterating
+| 'Reshuffle' >> beam.Reshuffle()
+| 'MakeKV' >> beam.Map(lambda x:
+ (b'', str(x).encode())).with_output_types(
+ typing.Tuple[bytes, bytes])
+| 'WriteToKafka' >> WriteToKafka(
+producer_config={'bootstrap.servers': self.bootstrap_servers},
+topic=self.topic,
+expansion_service=self.expansion_service))
+
+ def build_read_pipeline(self, pipeline):
+_ = (
+pipeline
+| 'ReadFromKafka' >> ReadFromKafka(
+consumer_config={
+'bootstrap.servers': self.bootstrap_servers,
+'auto.offset.reset': 'earliest'
+},
+topics=[self.topic],
+expansion_service=self.expansion_service)
+| 'Windowing' >> beam.WindowInto(
+beam.window.FixedWindows(300),
+trigger=beam.transforms.trigger.AfterProcessingTime(60),
+accumulation_mode=beam.transforms.trigger.AccumulationMode.
+DISCARDING)
+| 'DecodingValue' >> beam.Map(lambda elem: int(elem[1].decode()))
+| 'CombineGlobally' >> beam.CombineGlobally(sum).without_defaults()
+| 'SetSumCounter' >> beam.Map(self.sum_counter.inc))
+
+ def run_xlang_kafkaio(self, pipeline):
+self.build_write_pipeline(pipeline)
+self.build_read_pipeline(pipeline)
+pipeline.run(False)
+
+
+@unittest.skipUnless(
+os.environ.get('LOCAL_KAFKA_JAR'),
+"LOCAL_KAFKA_JAR environment var is not provided.")
+@unittest.skipUnless(
+os.environ.get('EXPANSION_JAR'),
+"EXPANSION_JAR environment var is not provided.")
+class CrossLanguageKafkaIOTest(unittest.TestCase):
+ def get_open_port(self):
+s = None
+try:
+ s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+except: # pylint: disable=bare-except
+ # Above call will fail for nodes that only support IPv6.
+ s = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
+s.bind(('localhost', 0))
+s.listen(1)
+port = s.getsockname()[1]
+s.close()
+return port
+
+ @contextlib.contextmanager
+ def local_services(self, expansion_service_jar_file, local_kafka_jar_file):
+expansion_service_port = str(self.get_open_port())
+kafka_port = str(self.get_open_port())
+zookeeper_port = str(self.get_open_port())
+
+expansion_server = None
+kafka_server = None
+try:
+ expansion_server = subprocess.Popen(
Review comment:
updated.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 438468)
Time Spent: 1h 10m (was: 1h)
> adding cross-language KafkaIO integration test
> --
>
> Key: BEAM-10125
> URL: https://issues.apache.org/jira/browse/BEAM-10125
> Project: Beam
>
[jira] [Work logged] (BEAM-10125) adding cross-language KafkaIO integration test
[
https://issues.apache.org/jira/browse/BEAM-10125?focusedWorklogId=438466=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-438466
]
ASF GitHub Bot logged work on BEAM-10125:
-
Author: ASF GitHub Bot
Created on: 28/May/20 21:47
Start Date: 28/May/20 21:47
Worklog Time Spent: 10m
Work Description: ihji commented on a change in pull request #11847:
URL: https://github.com/apache/beam/pull/11847#discussion_r432142695
##
File path: sdks/python/apache_beam/io/external/xlang_kafkaio_it_test.py
##
@@ -0,0 +1,145 @@
+"""Integration test for Python cross-language pipelines for Java KafkaIO."""
+
+from __future__ import absolute_import
+
+import contextlib
+import logging
+import os
+import socket
+import subprocess
+import time
+import typing
+import unittest
+
+import grpc
+
+import apache_beam as beam
+from apache_beam.io.external.kafka import ReadFromKafka
+from apache_beam.io.external.kafka import WriteToKafka
+from apache_beam.metrics import Metrics
+from apache_beam.options.pipeline_options import PipelineOptions
+from apache_beam.testing.test_pipeline import TestPipeline
+
+
+class CrossLanguageKafkaIO(object):
+ def __init__(self, bootstrap_servers, topic, expansion_service=None):
+self.bootstrap_servers = bootstrap_servers
+self.topic = topic
+self.expansion_service = expansion_service or (
+'localhost:%s' % os.environ.get('EXPANSION_PORT'))
+self.sum_counter = Metrics.counter('source', 'elements_sum')
+
+ def build_write_pipeline(self, pipeline):
+_ = (
+pipeline
+| 'Impulse' >> beam.Impulse()
+| 'Generate' >> beam.FlatMap(lambda x: range(1000)) # pylint:
disable=range-builtin-not-iterating
+| 'Reshuffle' >> beam.Reshuffle()
+| 'MakeKV' >> beam.Map(lambda x:
+ (b'', str(x).encode())).with_output_types(
+ typing.Tuple[bytes, bytes])
+| 'WriteToKafka' >> WriteToKafka(
+producer_config={'bootstrap.servers': self.bootstrap_servers},
+topic=self.topic,
+expansion_service=self.expansion_service))
+
+ def build_read_pipeline(self, pipeline):
+_ = (
+pipeline
+| 'ReadFromKafka' >> ReadFromKafka(
+consumer_config={
+'bootstrap.servers': self.bootstrap_servers,
+'auto.offset.reset': 'earliest'
+},
+topics=[self.topic],
+expansion_service=self.expansion_service)
+| 'Windowing' >> beam.WindowInto(
+beam.window.FixedWindows(300),
+trigger=beam.transforms.trigger.AfterProcessingTime(60),
+accumulation_mode=beam.transforms.trigger.AccumulationMode.
+DISCARDING)
+| 'DecodingValue' >> beam.Map(lambda elem: int(elem[1].decode()))
+| 'CombineGlobally' >> beam.CombineGlobally(sum).without_defaults()
+| 'SetSumCounter' >> beam.Map(self.sum_counter.inc))
+
+ def run_xlang_kafkaio(self, pipeline):
+self.build_write_pipeline(pipeline)
+self.build_read_pipeline(pipeline)
+pipeline.run(False)
+
+
+@unittest.skipUnless(
+os.environ.get('LOCAL_KAFKA_JAR'),
+"LOCAL_KAFKA_JAR environment var is not provided.")
+@unittest.skipUnless(
+os.environ.get('EXPANSION_JAR'),
+"EXPANSION_JAR environment var is not provided.")
+class CrossLanguageKafkaIOTest(unittest.TestCase):
+ def get_open_port(self):
+s = None
+try:
+ s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+except: # pylint: disable=bare-except
+ # Above call will fail for nodes that only support IPv6.
+ s = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
+s.bind(('localhost', 0))
+s.listen(1)
+port = s.getsockname()[1]
+s.close()
+return port
+
+ @contextlib.contextmanager
+ def local_services(self, expansion_service_jar_file, local_kafka_jar_file):
+expansion_service_port = str(self.get_open_port())
+kafka_port = str(self.get_open_port())
+zookeeper_port = str(self.get_open_port())
+
+expansion_server = None
+kafka_server = None
+try:
+ expansion_server = subprocess.Popen(
+ ['java', '-jar', expansion_service_jar_file, expansion_service_port])
+ kafka_server = subprocess.Popen(
Review comment:
Yes, this is for external testing (probably only for small scale
correctness tests, we may still need kubernetes cluster for large scale
performance tests).
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
---
Worklog Id: (was: 438466)
[jira] [Work logged] (BEAM-10125) adding cross-language KafkaIO integration test
[
https://issues.apache.org/jira/browse/BEAM-10125?focusedWorklogId=438467=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-438467
]
ASF GitHub Bot logged work on BEAM-10125:
-
Author: ASF GitHub Bot
Created on: 28/May/20 21:47
Start Date: 28/May/20 21:47
Worklog Time Spent: 10m
Work Description: ihji commented on a change in pull request #11847:
URL: https://github.com/apache/beam/pull/11847#discussion_r432142814
##
File path: sdks/python/apache_beam/io/external/xlang_kafkaio_it_test.py
##
@@ -0,0 +1,145 @@
+"""Integration test for Python cross-language pipelines for Java KafkaIO."""
+
+from __future__ import absolute_import
+
+import contextlib
+import logging
+import os
+import socket
+import subprocess
+import time
+import typing
+import unittest
+
+import grpc
+
+import apache_beam as beam
+from apache_beam.io.external.kafka import ReadFromKafka
+from apache_beam.io.external.kafka import WriteToKafka
+from apache_beam.metrics import Metrics
+from apache_beam.options.pipeline_options import PipelineOptions
+from apache_beam.testing.test_pipeline import TestPipeline
+
+
+class CrossLanguageKafkaIO(object):
+ def __init__(self, bootstrap_servers, topic, expansion_service=None):
+self.bootstrap_servers = bootstrap_servers
+self.topic = topic
+self.expansion_service = expansion_service or (
+'localhost:%s' % os.environ.get('EXPANSION_PORT'))
+self.sum_counter = Metrics.counter('source', 'elements_sum')
+
+ def build_write_pipeline(self, pipeline):
+_ = (
+pipeline
+| 'Impulse' >> beam.Impulse()
+| 'Generate' >> beam.FlatMap(lambda x: range(1000)) # pylint:
disable=range-builtin-not-iterating
+| 'Reshuffle' >> beam.Reshuffle()
+| 'MakeKV' >> beam.Map(lambda x:
+ (b'', str(x).encode())).with_output_types(
+ typing.Tuple[bytes, bytes])
+| 'WriteToKafka' >> WriteToKafka(
+producer_config={'bootstrap.servers': self.bootstrap_servers},
+topic=self.topic,
+expansion_service=self.expansion_service))
+
+ def build_read_pipeline(self, pipeline):
+_ = (
+pipeline
+| 'ReadFromKafka' >> ReadFromKafka(
+consumer_config={
+'bootstrap.servers': self.bootstrap_servers,
+'auto.offset.reset': 'earliest'
+},
+topics=[self.topic],
+expansion_service=self.expansion_service)
+| 'Windowing' >> beam.WindowInto(
+beam.window.FixedWindows(300),
+trigger=beam.transforms.trigger.AfterProcessingTime(60),
+accumulation_mode=beam.transforms.trigger.AccumulationMode.
+DISCARDING)
+| 'DecodingValue' >> beam.Map(lambda elem: int(elem[1].decode()))
+| 'CombineGlobally' >> beam.CombineGlobally(sum).without_defaults()
+| 'SetSumCounter' >> beam.Map(self.sum_counter.inc))
+
+ def run_xlang_kafkaio(self, pipeline):
+self.build_write_pipeline(pipeline)
+self.build_read_pipeline(pipeline)
+pipeline.run(False)
+
+
+@unittest.skipUnless(
+os.environ.get('LOCAL_KAFKA_JAR'),
+"LOCAL_KAFKA_JAR environment var is not provided.")
+@unittest.skipUnless(
+os.environ.get('EXPANSION_JAR'),
+"EXPANSION_JAR environment var is not provided.")
+class CrossLanguageKafkaIOTest(unittest.TestCase):
+ def get_open_port(self):
+s = None
+try:
+ s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+except: # pylint: disable=bare-except
+ # Above call will fail for nodes that only support IPv6.
+ s = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
+s.bind(('localhost', 0))
+s.listen(1)
+port = s.getsockname()[1]
+s.close()
+return port
+
+ @contextlib.contextmanager
+ def local_services(self, expansion_service_jar_file, local_kafka_jar_file):
+expansion_service_port = str(self.get_open_port())
+kafka_port = str(self.get_open_port())
+zookeeper_port = str(self.get_open_port())
+
+expansion_server = None
+kafka_server = None
+try:
+ expansion_server = subprocess.Popen(
+ ['java', '-jar', expansion_service_jar_file, expansion_service_port])
+ kafka_server = subprocess.Popen(
+ ['java', '-jar', local_kafka_jar_file, kafka_port, zookeeper_port])
+ time.sleep(3)
+ channel_creds = grpc.local_channel_credentials()
+ with grpc.secure_channel('localhost:%s' % expansion_service_port,
+ channel_creds) as channel:
+grpc.channel_ready_future(channel).result()
+
+ yield expansion_service_port, kafka_port
+finally:
+ if expansion_server:
+expansion_server.kill()
+ if kafka_server:
+kafka_server.kill()
+
+ def get_options(self):
+options = PipelineOptions([
+
[jira] [Work logged] (BEAM-10125) adding cross-language KafkaIO integration test
[
https://issues.apache.org/jira/browse/BEAM-10125?focusedWorklogId=438415=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-438415
]
ASF GitHub Bot logged work on BEAM-10125:
-
Author: ASF GitHub Bot
Created on: 28/May/20 20:07
Start Date: 28/May/20 20:07
Worklog Time Spent: 10m
Work Description: chamikaramj commented on a change in pull request
#11847:
URL: https://github.com/apache/beam/pull/11847#discussion_r432083261
##
File path: sdks/python/apache_beam/io/external/xlang_kafkaio_it_test.py
##
@@ -0,0 +1,145 @@
+"""Integration test for Python cross-language pipelines for Java KafkaIO."""
+
+from __future__ import absolute_import
+
+import contextlib
+import logging
+import os
+import socket
+import subprocess
+import time
+import typing
+import unittest
+
+import grpc
+
+import apache_beam as beam
+from apache_beam.io.external.kafka import ReadFromKafka
+from apache_beam.io.external.kafka import WriteToKafka
+from apache_beam.metrics import Metrics
+from apache_beam.options.pipeline_options import PipelineOptions
+from apache_beam.testing.test_pipeline import TestPipeline
+
+
+class CrossLanguageKafkaIO(object):
+ def __init__(self, bootstrap_servers, topic, expansion_service=None):
+self.bootstrap_servers = bootstrap_servers
+self.topic = topic
+self.expansion_service = expansion_service or (
+'localhost:%s' % os.environ.get('EXPANSION_PORT'))
+self.sum_counter = Metrics.counter('source', 'elements_sum')
+
+ def build_write_pipeline(self, pipeline):
+_ = (
+pipeline
+| 'Impulse' >> beam.Impulse()
+| 'Generate' >> beam.FlatMap(lambda x: range(1000)) # pylint:
disable=range-builtin-not-iterating
+| 'Reshuffle' >> beam.Reshuffle()
+| 'MakeKV' >> beam.Map(lambda x:
+ (b'', str(x).encode())).with_output_types(
+ typing.Tuple[bytes, bytes])
+| 'WriteToKafka' >> WriteToKafka(
+producer_config={'bootstrap.servers': self.bootstrap_servers},
+topic=self.topic,
+expansion_service=self.expansion_service))
+
+ def build_read_pipeline(self, pipeline):
+_ = (
+pipeline
+| 'ReadFromKafka' >> ReadFromKafka(
+consumer_config={
+'bootstrap.servers': self.bootstrap_servers,
+'auto.offset.reset': 'earliest'
+},
+topics=[self.topic],
+expansion_service=self.expansion_service)
+| 'Windowing' >> beam.WindowInto(
+beam.window.FixedWindows(300),
+trigger=beam.transforms.trigger.AfterProcessingTime(60),
+accumulation_mode=beam.transforms.trigger.AccumulationMode.
+DISCARDING)
+| 'DecodingValue' >> beam.Map(lambda elem: int(elem[1].decode()))
+| 'CombineGlobally' >> beam.CombineGlobally(sum).without_defaults()
+| 'SetSumCounter' >> beam.Map(self.sum_counter.inc))
+
+ def run_xlang_kafkaio(self, pipeline):
+self.build_write_pipeline(pipeline)
+self.build_read_pipeline(pipeline)
+pipeline.run(False)
+
+
+@unittest.skipUnless(
+os.environ.get('LOCAL_KAFKA_JAR'),
+"LOCAL_KAFKA_JAR environment var is not provided.")
+@unittest.skipUnless(
+os.environ.get('EXPANSION_JAR'),
+"EXPANSION_JAR environment var is not provided.")
+class CrossLanguageKafkaIOTest(unittest.TestCase):
+ def get_open_port(self):
+s = None
+try:
+ s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+except: # pylint: disable=bare-except
+ # Above call will fail for nodes that only support IPv6.
+ s = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
+s.bind(('localhost', 0))
+s.listen(1)
+port = s.getsockname()[1]
+s.close()
+return port
+
+ @contextlib.contextmanager
+ def local_services(self, expansion_service_jar_file, local_kafka_jar_file):
+expansion_service_port = str(self.get_open_port())
+kafka_port = str(self.get_open_port())
+zookeeper_port = str(self.get_open_port())
+
+expansion_server = None
+kafka_server = None
+try:
+ expansion_server = subprocess.Popen(
Review comment:
Is this only for internal testing ?
Externally, kafkaio.py should automatically startup an expansion service as
long as we are in a release or a Beam repo clone.
##
File path: sdks/python/apache_beam/io/external/xlang_kafkaio_it_test.py
##
@@ -0,0 +1,145 @@
+"""Integration test for Python cross-language pipelines for Java KafkaIO."""
+
+from __future__ import absolute_import
+
+import contextlib
+import logging
+import os
+import socket
+import subprocess
+import time
+import typing
+import unittest
+
+import grpc
+
+import apache_beam as beam
+from apache_beam.io.external.kafka import ReadFromKafka
+from apache_beam.io.external.kafka import
[jira] [Work logged] (BEAM-10125) adding cross-language KafkaIO integration test
[ https://issues.apache.org/jira/browse/BEAM-10125?focusedWorklogId=438223=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-438223 ] ASF GitHub Bot logged work on BEAM-10125: - Author: ASF GitHub Bot Created on: 28/May/20 08:04 Start Date: 28/May/20 08:04 Worklog Time Spent: 10m Work Description: ihji commented on pull request #11847: URL: https://github.com/apache/beam/pull/11847#issuecomment-635184781 This change is also used for XLang KafkaIO internal test. This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 438223) Time Spent: 0.5h (was: 20m) > adding cross-language KafkaIO integration test > -- > > Key: BEAM-10125 > URL: https://issues.apache.org/jira/browse/BEAM-10125 > Project: Beam > Issue Type: Improvement > Components: io-java-kafka >Reporter: Heejong Lee >Assignee: Heejong Lee >Priority: P2 > Time Spent: 0.5h > Remaining Estimate: 0h > > adding cross-language KafkaIO integration test -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-10125) adding cross-language KafkaIO integration test
[ https://issues.apache.org/jira/browse/BEAM-10125?focusedWorklogId=438222=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-438222 ] ASF GitHub Bot logged work on BEAM-10125: - Author: ASF GitHub Bot Created on: 28/May/20 08:03 Start Date: 28/May/20 08:03 Worklog Time Spent: 10m Work Description: ihji commented on pull request #11847: URL: https://github.com/apache/beam/pull/11847#issuecomment-635184129 R: @chamikaramj This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 438222) Time Spent: 20m (was: 10m) > adding cross-language KafkaIO integration test > -- > > Key: BEAM-10125 > URL: https://issues.apache.org/jira/browse/BEAM-10125 > Project: Beam > Issue Type: Improvement > Components: io-java-kafka >Reporter: Heejong Lee >Assignee: Heejong Lee >Priority: P2 > Time Spent: 20m > Remaining Estimate: 0h > > adding cross-language KafkaIO integration test -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Work logged] (BEAM-10125) adding cross-language KafkaIO integration test
[ https://issues.apache.org/jira/browse/BEAM-10125?focusedWorklogId=438220=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-438220 ] ASF GitHub Bot logged work on BEAM-10125: - Author: ASF GitHub Bot Created on: 28/May/20 08:01 Start Date: 28/May/20 08:01 Worklog Time Spent: 10m Work Description: ihji opened a new pull request #11847: URL: https://github.com/apache/beam/pull/11847 Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily: - [ ] [**Choose reviewer(s)**](https://beam.apache.org/contribute/#make-your-change) and mention them in a comment (`R: @username`). - [ ] Format the pull request title like `[BEAM-XXX] Fixes bug in ApproximateQuantiles`, where you replace `BEAM-XXX` with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue. - [ ] Update `CHANGES.md` with noteworthy changes. - [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf). See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://beam.apache.org/contribute/#make-reviewers-job-easier). Post-Commit Tests Status (on master branch) Lang | SDK | Apex | Dataflow | Flink | Gearpump | Samza | Spark --- | --- | --- | --- | --- | --- | --- | --- Go | [](https://builds.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/) | --- | --- | [](https://builds.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/) Java | [](https://builds.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Apex/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/)[](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/)[](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Java11/lastCompletedBuild/)[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/)[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Gearpump/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/)[](https://builds.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/)[