1 files changed, 35 insertions, 0 deletions
diff --git a/charms/trusty/kafka/actions/read-topic b/charms/trusty/kafka/actions/read-topic
new file mode 100755
index 0000000..9f59396
--- /dev/null
+++ b/charms/trusty/kafka/actions/read-topic
@@ -0,0 +1,35 @@
+#!/usr/bin/env python
+#pylint: disable=C0103
+try:
+    from charmhelpers.core import hookenv
+    from charmhelpers.core import unitdata
+    import jujubigdata
+    from jujubigdata import utils
+    charm_ready = unitdata.kv().get('charm.active', False)
+except ImportError:
+    charm_ready = False
+
+if not charm_ready:
+    # might not have hookenv.action_fail available yet
+    from subprocess import call
+    call(['action-fail', 'Kafka service not yet ready'])
+
+kafka_reqs = ['vendor', 'packages', 'groups', 'users', 'dirs', 'ports']
+dist_config = jujubigdata.utils.DistConfig(filename='dist.yaml',
+                                           required_keys=kafka_reqs)
+
+# Grab the business
+topic_name = hookenv.action_get('topic')
+topic_partition = hookenv.action_get('partition')
+
+output = utils.run_as(
+    'kafka', 'kafka-simple-consumer-shell.sh',
+    '--broker-list', '{}:{}'.format(
+        hookenv.unit_private_ip(),
+        dist_config.port('kafka'),
+    ),
+    '--topic', topic_name,
+    '--partition', topic_partition,
+    '--no-wait-at-logend',
+    capture_output=True)
+hookenv.action_set({'output': output})