|
|
@@ -19,6 +19,8 @@
|
|
|
import logging
|
|
|
import json
|
|
|
|
|
|
+from subprocess import call
|
|
|
+
|
|
|
from django.core.cache import cache
|
|
|
from django.utils.translation import ugettext as _
|
|
|
|
|
|
@@ -27,6 +29,7 @@ from desktop.lib.rest.resource import Resource
|
|
|
from desktop.lib.i18n import smart_unicode
|
|
|
|
|
|
from kafka.conf import KAFKA
|
|
|
+from libzookeeper.conf import zkensemble
|
|
|
|
|
|
|
|
|
LOG = logging.getLogger(__name__)
|
|
|
@@ -46,8 +49,6 @@ class KafkaApiException(Exception):
|
|
|
class KafkaApi(object):
|
|
|
"""
|
|
|
https://github.com/confluentinc/kafka-rest
|
|
|
-
|
|
|
- create/delete topics are not available in the REST API.
|
|
|
"""
|
|
|
|
|
|
def __init__(self, user=None, security_enabled=False, ssl_cert_ca_verify=False):
|
|
|
@@ -64,3 +65,17 @@ class KafkaApi(object):
|
|
|
return json.loads(response)
|
|
|
except RestException, e:
|
|
|
raise KafkaApiException(e)
|
|
|
+
|
|
|
+ def create_topic(self, name, partitions=1, replication_factor=1):
|
|
|
+ # Create/delete topics are not available in the REST API.
|
|
|
+ # Here only works with hack if command is available on the Hue host.
|
|
|
+ try:
|
|
|
+ return call(
|
|
|
+ 'kafka-topics --zookeeper %(zookeeper)s --create --if-not-exists --topic %(name)s --partitions %(partitions)s --replication-factor %(replication_factor)s' % {
|
|
|
+ 'zookeeper': zkensemble(),
|
|
|
+ 'name': name,
|
|
|
+ 'partitions': partitions,
|
|
|
+ 'replication_factor': replication_factor
|
|
|
+ })
|
|
|
+ except RestException, e:
|
|
|
+ raise KafkaApiException(e)
|