Ver código fonte

HUE-8553 [kafka] Add a workaround API for creating a topic

Romain Rigaux 7 anos atrás
pai
commit
74f835c

+ 18 - 0
desktop/libs/kafka/src/kafka/kafka_api.py

@@ -80,6 +80,24 @@ def list_topic(request):
   })
 
 
+@error_handler
+def create_topic(request):
+  name = request.POST.get('name')
+  partitions = request.POST.get('partitions', 1)
+  replication_factor = request.POST.get('replication_factor', 1)
+
+  status = KafkaApi().create_topic(name, partitions, replication_factor)
+
+  return JsonResponse({
+    'status': status,
+    'topic': {
+      'name': name,
+      'partitions': partitions,
+      'replication_factor': replication_factor
+    }
+  })
+
+
 def get_topics():
   if has_kafka_api():
     return KafkaApi().topics()

+ 17 - 2
desktop/libs/kafka/src/kafka/kafka_client.py

@@ -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)

+ 1 - 0
desktop/libs/kafka/src/kafka/urls.py

@@ -21,4 +21,5 @@ from kafka import kafka_api as kafka_kafka_api
 urlpatterns = [
   url(r'^api/topics/list/$', kafka_kafka_api.list_topics, name='list_topics'),
   url(r'^api/topic/list/$', kafka_kafka_api.list_topic, name='list_topic'),
+  url(r'^api/topic/create/$', kafka_kafka_api.create_topic, name='create_topic'),
 ]

+ 1 - 0
desktop/libs/notebook/src/notebook/api.py

@@ -166,6 +166,7 @@ def execute(request, engine=None):
   snippet = json.loads(request.POST.get('snippet', '{}'))
 
   response = _execute_notebook(request, notebook, snippet)
+  print response
 
   return JsonResponse(response)
 

+ 0 - 2
desktop/libs/notebook/src/notebook/connectors/rdbms.py

@@ -15,10 +15,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import json
 import logging
 
-from desktop.lib import export_csvxls
 from desktop.lib.i18n import force_unicode
 
 from beeswax import data_export