replace confluent with samrama (#306)

CMakeLists.txt

@@ -202,10 +202,6 @@ add_library(algorithm SHARED
 
 target_link_libraries(algorithm mem)
 ################################
-# rdkafka
-################################
-include(rdkafka.cmake)
-################################
 # Unit Tests
 ################################
 file(GLOB QUERY_UNITTEST_FILES query/*_unittest.cu)
@@ -255,13 +251,11 @@ execute_process(
 )
 
 add_custom_target(lint-all
-        DEPENDS rdkafka::rdkafka lib
         COMMAND ./scripts/clang-lint.sh ${ALL_C_SRC}
         COMMAND ./scripts/golang-lint.sh ${ALL_GO_SRC}
         VERBATIM)
 
 add_custom_target(lint
-        DEPENDS rdkafka::rdkafka lib
         COMMAND ./scripts/clang-lint.sh ${CHANGED_C_SRC}
         COMMAND ./scripts/golang-lint.sh ${CHANGED_GO_SRC}
         VERBATIM)
@@ -280,7 +274,7 @@ add_custom_target(aresbrokerd DEPENDS ${ALL_GO_SRC_LIST}
         VERBATIM 
         )
 
-add_custom_target(ares-subscriber DEPENDS rdkafka::rdkafka ${ALL_GO_SRC_LIST}
+add_custom_target(ares-subscriber DEPENDS ${ALL_GO_SRC_LIST}
         COMMAND go build -tags static -o bin/ares-subscriber ./cmd/subscriber/main.go
         VERBATIM
         )
@@ -289,15 +283,15 @@ add_custom_target(run_server DEPENDS aresd
         COMMAND bash -c 'DYLD_LIBRARY_PATH=$$LIBRARY_PATH ./bin/aresd'
         )
 
-add_custom_target(test-golang DEPENDS aresd rdkafka::rdkafka
+add_custom_target(test-golang
         COMMAND bash -c 'ARES_ENV=test DYLD_LIBRARY_PATH=$$LIBRARY_PATH ginkgo -r'
         )
 
 ###############################
 # misc
 ###############################
 
-add_custom_target(travis DEPENDS rdkafka::rdkafka
+add_custom_target(travis
         COMMAND bash -c 'ARES_ENV=test .travis/run_unittest.sh'
         )
 

controller/models/job.go

@@ -44,11 +44,11 @@ type KafkaConfig struct {
 	RestartInterval     int            `json:"restartInterval,omitempty"`
 	FailureHandler      FailureHandler `json:"failureHandler,omitempty"`
 
-	// confluent kafka
-	KafkaBroker       string `json:"kafkaBroker" yaml:"kafkaBroker"`
-	MaxPollIntervalMs int    `json:"maxPollIntervalMs" yaml:"maxPollIntervalMs" default:"300000"`
-	SessionTimeoutNs  int    `json:"sessionTimeoutNs" yaml:"sessionTimeoutNs" default:"10000"`
-	ChannelBufferSize uint   `json:"channelBufferSize" yaml:"channelBufferSize" default:"256"`
+	// sarama config
+	KafkaBroker        string `json:"kafkaBroker" yaml:"kafkaBroker"`
+	SessionTimeoutMs   int    `json:"sessionTimeoutMs" yaml:"sessionTimeoutMs" default:"10000"`
+	ChannelBufferSize  uint   `json:"channelBufferSize" yaml:"channelBufferSize" default:"256"`
+	ReblanceTimeoutSec int    `json:"reblanceTimeoutSec" yaml:"reblanceTimeoutSec" default:"10"`
 }
 
 // JobConfig is job's config

controller/tasks/etcd/ingestion_assignment_test.go

@@ -311,9 +311,9 @@ func TestIngestionAssignmentTask(t *testing.T) {
         }
       },
       "kafkaBroker": "",
-      "maxPollIntervalMs": 0,
-      "sessionTimeoutNs": 0,
-      "channelBufferSize": 0
+      "sessionTimeoutMs": 0,
+      "channelBufferSize": 0,
+      "reblanceTimeoutSec": 0
     }
   }
 ]`