Optional kafka (#1355)

* Zookeeper include fix

* Update submodule contrib/poco

* Fix zookeeper submodule ho-to-make doc

* Update zookeeper submodule

* Update submodule zookeeper

* Fix compile with external zookeeper

* Optional kafka

* fix

* Fix

* fix

cmake/find_rdkafka.cmake

+option (ENABLE_RDKAFKA "Enable kafka" ON)
+
+if (ENABLE_RDKAFKA)
+
 option (USE_INTERNAL_RDKAFKA_LIBRARY "Set to FALSE to use system librdkafka instead of the bundled" ${NOT_UNBUNDLED})
 
 if (NOT USE_INTERNAL_RDKAFKA_LIBRARY)
@@ -11,7 +15,10 @@ else ()
     set (USE_INTERNAL_RDKAFKA_LIBRARY 1)
     set (RDKAFKA_INCLUDE_DIR "${ClickHouse_SOURCE_DIR}/contrib/librdkafka/src")
     set (RDKAFKA_LIBRARY rdkafka)
-    set (USE_RDKAFKA 1)
 endif ()
 
-message (STATUS "Using librdkafka: ${RDKAFKA_INCLUDE_DIR} : ${RDKAFKA_LIBRARY}")
+set (USE_RDKAFKA 1)
+
+endif ()
+
+message (STATUS "Using librdkafka=${USE_RDKAFKA}: ${RDKAFKA_INCLUDE_DIR} : ${RDKAFKA_LIBRARY}")

cmake/find_zookeeper.cmake

@@ -3,6 +3,7 @@ option (USE_INTERNAL_ZOOKEEPER_LIBRARY "Set to FALSE to use system zookeeper lib
 if (NOT USE_INTERNAL_ZOOKEEPER_LIBRARY)
     find_library (ZOOKEEPER_LIBRARY zookeeper_mt)
     find_path (ZOOKEEPER_INCLUDE_DIR NAMES zookeeper/zookeeper.h PATHS ${ZOOKEEPER_INCLUDE_PATHS})
+    set(ZOOKEEPER_INCLUDE_DIR "${ZOOKEEPER_INCLUDE_DIR}/zookeeper")
 endif ()
 
 if (ZOOKEEPER_LIBRARY AND ZOOKEEPER_INCLUDE_DIR)

contrib/CMakeLists.txt

@@ -4,7 +4,6 @@ if (USE_INTERNAL_BOOST_LIBRARY)
     add_subdirectory (libboost)
 endif ()
 
-
 if (USE_INTERNAL_LZ4_LIBRARY)
     add_subdirectory (lz4-cmake)
 endif ()
@@ -62,8 +61,8 @@ if (NOT ARCH_ARM)
 endif ()
 
 if (USE_INTERNAL_RDKAFKA_LIBRARY)
-    set(RDKAFKA_BUILD_EXAMPLES OFF)
-    set(RDKAFKA_BUILD_TESTS OFF)
+    set(RDKAFKA_BUILD_EXAMPLES OFF CACHE BOOL "")
+    set(RDKAFKA_BUILD_TESTS OFF CACHE BOOL "")
     mark_as_advanced(ZLIB_INCLUDE_DIR)
     add_subdirectory (librdkafka)
 endif ()

dbms/CMakeLists.txt

@@ -181,7 +181,6 @@ target_link_libraries (dbms
     ${Boost_SYSTEM_LIBRARY}
     ${Poco_Data_LIBRARY}
     ${BTRIE_LIBRARIES}
-    ${RDKAFKA_LIBRARY}
 )
 
 if (Poco_DataODBC_FOUND)
@@ -220,7 +219,10 @@ endif ()
 
 target_include_directories (dbms PUBLIC ${DBMS_INCLUDE_DIR})
 target_include_directories (dbms PUBLIC ${PCG_RANDOM_INCLUDE_DIR})
-target_include_directories (dbms PUBLIC ${RDKAFKA_INCLUDE_DIR})
+
+if (USE_RDKAFKA)
+    target_link_libraries (dbms ${RDKAFKA_LIBRARY})
+endif ()
 
 # only for copy_headers.sh:
 target_include_directories (dbms PRIVATE ${COMMON_INCLUDE_DIR})

dbms/src/Common/config.h.in

@@ -6,6 +6,7 @@
 #cmakedefine01 USE_MYSQL
 #cmakedefine01 USE_RE2_ST
 #cmakedefine01 USE_VECTORCLASS
+#cmakedefine01 USE_RDKAFKA
 #cmakedefine01 Poco_DataODBC_FOUND
 #cmakedefine01 Poco_MongoDB_FOUND
 #cmakedefine01 Poco_NetSSL_FOUND

dbms/src/Storages/StorageFactory.cpp

+#include <unistd.h>
 #include <Poco/Util/Application.h>
 #include <Poco/Util/AbstractConfiguration.h>
-
 #include <Core/FieldVisitors.h>
 #include <Common/StringUtils.h>
 #include <Common/typeid_cast.h>
-
 #include <DataTypes/DataTypeTuple.h>
-
 #include <Parsers/ASTCreateQuery.h>
 #include <Parsers/ASTFunction.h>
 #include <Parsers/ASTIdentifier.h>
 #include <Parsers/ASTLiteral.h>
-
 #include <Interpreters/Context.h>
 #include <Interpreters/evaluateConstantExpression.h>
 #include <Interpreters/ExpressionAnalyzer.h>
 #include <Interpreters/getClusterName.h>
-
 #include <Storages/StorageLog.h>
 #include <Storages/StorageTinyLog.h>
 #include <Storages/StorageStripeLog.h>
@@ -35,11 +31,13 @@
 #include <Storages/StorageJoin.h>
 #include <Storages/StorageFile.h>
 #include <Storages/StorageDictionary.h>
-#include <Storages/StorageKafka.h>
 #include <AggregateFunctions/AggregateFunctionFactory.h>
 #include <AggregateFunctions/parseAggregateFunctionParameters.h>
 
-#include <unistd.h>
+#include <Common/config.h>
+#if USE_RDKAFKA
+#include <Storages/StorageKafka.h>
+#endif
 
 
 namespace DB
@@ -59,6 +57,7 @@ namespace ErrorCodes
     extern const int TYPE_MISMATCH;
     extern const int INCORRECT_NUMBER_OF_COLUMNS;
     extern const int DATA_TYPE_CANNOT_BE_USED_IN_TABLES;
+    extern const int SUPPORT_IS_DISABLED;
 }
 
 
@@ -628,6 +627,7 @@ StoragePtr StorageFactory::get(
     }
     else if (name == "Kafka")
     {
+#if USE_RDKAFKA
         /** Arguments of engine is following:
           * - Kafka broker list
           * - List of topics
@@ -684,6 +684,9 @@ StoragePtr StorageFactory::get(
             table_name, database_name, context, columns,
             materialized_columns, alias_columns, column_defaults,
             brokers, group, topics, format, schema);
+#else
+            throw Exception{"Storage `Kafka` disabled because ClickHouse built without kafka support.", ErrorCodes::SUPPORT_IS_DISABLED};
+#endif
     }
     else if (endsWith(name, "MergeTree"))
     {

dbms/src/Storages/StorageKafka.cpp

+#include <Common/config.h>
+#if USE_RDKAFKA
+
+#include <thread>
 #include <Common/Exception.h>
 #include <Common/setThreadName.h>
 #include <DataStreams/FormatFactory.h>
@@ -14,8 +18,6 @@
 
 #include <rdkafka.h>
 
-#include <thread>
-
 namespace DB
 {
 
@@ -356,3 +358,5 @@ void StorageKafka::streamToViews()
 
 
 }
+
+#endif

dbms/src/Storages/StorageKafka.h

 #pragma once
+#include <Common/config.h>
+#if USE_RDKAFKA
 
 #include <mutex>
 
@@ -85,3 +87,5 @@ private:
 };
 
 }
+
+#endif