Merge pull request #139 from conker84/3.5

Rebase from v3.4 pre-release

Author: jexp

consumer/src/main/kotlin/streams/StreamsEventSink.kt

@@ -26,7 +26,7 @@ abstract class StreamsEventConsumer<T>(private val consumer: T, config: StreamsS
 
     abstract fun start()
 
-    abstract fun read(): Map<String, List<Map<String, Any?>>>?
+    abstract fun read(): Map<String, List<Any>>?
 
 }
 

consumer/src/main/kotlin/streams/StreamsEventSinkQueryExecution.kt

@@ -1,6 +1,5 @@
 package streams
 
-import org.neo4j.graphdb.GraphDatabaseService
 import org.neo4j.kernel.internal.GraphDatabaseAPI
 import org.neo4j.logging.Log
 import streams.utils.Neo4jUtils
@@ -14,12 +13,16 @@ class StreamsEventSinkQueryExecution(private val streamsTopicService: StreamsTop
             return
         }
         val query = "${StreamsUtils.UNWIND} $cypherQuery"
-        if(log.isDebugEnabled){
-            log.debug("Processing ${params.size} events with query: $query")
+        if (log.isDebugEnabled) {
+            log.debug("Processing ${params.size} events, for topic $topic with query: $query")
         }
         if (Neo4jUtils.isWriteableInstance(db)) {
             try {
-                db.execute(query, mapOf("events" to params)).close()
+                val result = db.execute(query, mapOf("events" to params))
+                if (log.isDebugEnabled) {
+                    log.debug("Query statistics:\n${result.queryStatistics}")
+                }
+                result.close()
             } catch (e: Exception) {
                 log.error("Error while executing the query", e)
             }

consumer/src/main/kotlin/streams/StreamsSinkConfiguration.kt

@@ -13,7 +13,7 @@ private object StreamsSinkConfigurationConstants {
 
 data class StreamsSinkConfiguration(val enabled: Boolean = true,
                                     val proceduresEnabled: Boolean = true,
-                                    val sinkPollingInterval: Long = Long.MAX_VALUE,
+                                    val sinkPollingInterval: Long = 10000,
                                     val topics: Map<String, String> = emptyMap()) {
 
     companion object {

consumer/src/main/kotlin/streams/StreamsTopicService.kt

@@ -19,6 +19,9 @@ class StreamsTopicService(private val db: GraphDatabaseAPI, private val topicMap
     }
 
     fun clear() {
+        if (!Neo4jUtils.isWriteableInstance(db)) {
+            return
+        }
         return db.beginTx().use {
             val keys = properties.allProperties
                     .filterKeys { it.startsWith(STREAMS_TOPIC_KEY) }
@@ -31,6 +34,9 @@ class StreamsTopicService(private val db: GraphDatabaseAPI, private val topicMap
     }
 
     fun remove(topic: String) {
+        if (!Neo4jUtils.isWriteableInstance(db)) {
+            return
+        }
         val key = "$STREAMS_TOPIC_KEY$topic"
         return db.beginTx().use {
             if (!properties.hasProperty(key)) {

consumer/src/main/kotlin/streams/kafka/KafkaEventSink.kt

@@ -1,11 +1,7 @@
 package streams.kafka
 
-import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.GlobalScope
-import kotlinx.coroutines.Job
-import kotlinx.coroutines.launch
+import kotlinx.coroutines.*
 import org.apache.kafka.clients.consumer.KafkaConsumer
-import org.apache.kafka.common.errors.WakeupException
 import org.neo4j.kernel.configuration.Config
 import org.neo4j.logging.Log
 import streams.*
@@ -49,17 +45,15 @@ class KafkaEventSink(private val config: Config,
                 .createStreamsEventConsumer(config.raw, log)
                 .withTopics(topics.keys)
         this.eventConsumer.start()
-        if (log.isDebugEnabled) {
-            log.debug("Subscribed topics with queries: $topics")
-        }
         this.job = createJob()
         log.info("Kafka Sink started")
     }
 
-    override fun stop() {
+    override fun stop() = runBlocking {
         log.info("Stopping Sink daemon Job")
-        this.eventConsumer.stop()
-        StreamsUtils.ignoreExceptions({ job.cancel() }, UninitializedPropertyAccessException::class.java)
+        try {
+            job.cancelAndJoin()
+        } catch (e : UninitializedPropertyAccessException) { /* ignoring this one only */ }
     }
 
     override fun getEventSinkConfigMapper(): StreamsEventSinkConfigMapper {
@@ -78,7 +72,7 @@ class KafkaEventSink(private val config: Config,
         log.info("Creating Sink daemon Job")
         return GlobalScope.launch(Dispatchers.IO) {
             try {
-                while (true) {
+                while (isActive) {
                     val data= eventConsumer.read()
                     data?.forEach {
                         if (log.isDebugEnabled) {
@@ -87,6 +81,7 @@ class KafkaEventSink(private val config: Config,
                         queryExecution.execute(it.key, it.value)
                     }
                 }
+                eventConsumer.stop()
             } catch (e: Throwable) {
                 val message = e.message ?: "Generic error, please check the stack trace: "
                 log.error(message, e)
@@ -114,20 +109,18 @@ class KafkaEventConsumer(private val consumer: KafkaConsumer<String, ByteArray>,
             return
         }
         this.consumer.subscribe(topics)
-        log.info("Subscribing topics: $topics")
     }
 
     override fun stop() {
         StreamsUtils.ignoreExceptions({ consumer.close() }, UninitializedPropertyAccessException::class.java)
     }
 
-    override fun read(): Map<String, List<Map<String, Any?>>>? {
+    override fun read(): Map<String, List<Any>>? {
         val records = consumer.poll(config.sinkPollingInterval)
         if (records != null && !records.isEmpty) {
             return records
                     .map {
-                        it.topic()!! to JSONUtils.readValue(it.value(), Map::class.java)
-                                .mapKeys { it.key.toString() }
+                        it.topic()!! to JSONUtils.readValue(it.value(), Any::class.java)
                     }
                     .groupBy({ it.first }, { it.second })
         }

consumer/src/main/kotlin/streams/procedures/StreamsSinkProcedures.kt

@@ -1,22 +1,21 @@
 package streams.procedures
 
-import org.apache.commons.lang3.StringUtils
 import org.neo4j.logging.Log
 import org.neo4j.procedure.*
 import streams.StreamsEventConsumerFactory
 import streams.StreamsEventSinkConfigMapper
 import streams.StreamsSinkConfiguration
 import java.util.stream.Stream
 
-class StreamResult(@JvmField val event: Any)
+class StreamResult(@JvmField val event: Map<String, *>)
 
 class StreamsSinkProcedures {
 
     @JvmField @Context
     var log: Log? = null
 
     @Procedure(mode = Mode.SCHEMA, name = "streams.consume")
-    @Description("streams.consume(topic, config) - Allows to subscribe custom topics")
+    @Description("streams.consume(topic, {timeout: <long value>, from: <string>}) YIELD event - Allows to consume custom topics")
     fun consume(@Name("topic") topic: String?,
                 @Name(value = "config", defaultValue = "{}") config: Map<String, Any>?): Stream<StreamResult> {
         checkEnabled()
@@ -33,13 +32,20 @@ class StreamsSinkProcedures {
                 .createStreamsEventConsumer(configuration, log!!)
                 .withTopics(setOf(topic))
         consumer.start()
-        val data = consumer.read()
+        val data = try {
+            consumer.read()
+        } catch (e: Exception) {
+            if (log?.isDebugEnabled!!) {
+                log?.error("Error while consuming data", e)
+            }
+            emptyMap<String, List<Any>>()
+        }
         consumer.stop()
 
         if (log?.isDebugEnabled!!) {
-            log?.debug("Data retrieved after ${configuration["streams.sink.polling.interval"]} milliseconds: $data")
+            log?.debug("Data retrieved from topic $topic after ${configuration["streams.sink.polling.interval"]} milliseconds: $data")
         }
-        return data?.values?.flatMap { list -> list.map { StreamResult(it) } }?.stream() ?: Stream.empty()
+        return data?.values?.flatMap { list -> list.map { StreamResult(mapOf("data" to it)) } }?.stream() ?: Stream.empty()
     }
 
     private fun checkEnabled() {

consumer/src/test/kotlin/integrations/StreamsSinkProceduresIT.kt

@@ -44,8 +44,6 @@ class StreamsSinkProceduresIT {
 
     private val cypherQueryTemplate = "MERGE (n:Label {id: event.id}) ON CREATE SET n += event.properties"
 
-    private val topics = listOf("shouldWriteCypherQuery")
-
     @Rule
     @JvmField
     var testName = TestName()
@@ -114,7 +112,8 @@ class StreamsSinkProceduresIT {
         assertTrue { resultMap.containsKey("event") }
         assertNotNull(resultMap["event"], "should contain event")
         val event = resultMap["event"] as Map<String, Any?>
-        assertEquals(data, event)
+        val resultData = event["data"] as Map<String, Any?>
+        assertEquals(data, resultData)
     }
 
     @Test
@@ -123,4 +122,51 @@ class StreamsSinkProceduresIT {
         assertFalse { result.hasNext() }
     }
 
+    @Test
+    fun shouldReadArrayOfJson() {
+        val topic = "array-topic"
+        val list = listOf(data, data)
+        val producerRecord = ProducerRecord(topic, UUID.randomUUID().toString(), JSONUtils.writeValueAsBytes(list))
+        kafkaProducer.send(producerRecord).get()
+        db.execute("""
+            CALL streams.consume('$topic', {timeout: 5000}) YIELD event
+            UNWIND event.data AS data
+            CREATE (t:TEST) SET t += data.properties
+        """.trimIndent()).close()
+        val searchResult = db.execute("MATCH (t:TEST) WHERE properties(t) = {props} RETURN count(t) AS count", mapOf("props" to dataProperties))
+        assertTrue { searchResult.hasNext() }
+        val searchResultMap = searchResult.next()
+        assertTrue { searchResultMap.containsKey("count") }
+        assertEquals(2L, searchResultMap["count"])
+    }
+
+    @Test
+    fun shouldReadSimpleDataType() {
+        val topic = "simple-data"
+        val simpleInt = 1
+        val simpleBoolean = true
+        val simpleString = "test"
+        var producerRecord = ProducerRecord(topic, UUID.randomUUID().toString(), JSONUtils.writeValueAsBytes(simpleInt))
+        kafkaProducer.send(producerRecord).get()
+        producerRecord = ProducerRecord(topic, UUID.randomUUID().toString(), JSONUtils.writeValueAsBytes(simpleBoolean))
+        kafkaProducer.send(producerRecord).get()
+        producerRecord = ProducerRecord(topic, UUID.randomUUID().toString(), JSONUtils.writeValueAsBytes(simpleString))
+        kafkaProducer.send(producerRecord).get()
+        db.execute("""
+            CALL streams.consume('$topic', {timeout: 5000}) YIELD event
+            MERGE (t:LOG{simpleData: event.data})
+            RETURN count(t) AS insert
+        """.trimIndent()).close()
+        val searchResult = db.execute("""
+            MATCH (l:LOG)
+            WHERE l.simpleData IN [$simpleInt, $simpleBoolean, "$simpleString"]
+            RETURN count(l) as count
+        """.trimIndent())
+        assertTrue { searchResult.hasNext() }
+        val searchResultMap = searchResult.next()
+        assertTrue { searchResultMap.containsKey("count") }
+        assertEquals(3L, searchResultMap["count"])
+
+    }
+
 }

consumer/src/test/kotlin/streams/StreamsSinkConfigurationTest.kt

@@ -33,7 +33,7 @@ class StreamsSinkConfigurationTest {
     companion object {
         fun testDefaultConf(default: StreamsSinkConfiguration) {
             assertEquals(emptyMap(), default.topics)
-            assertEquals(Long.MAX_VALUE, default.sinkPollingInterval)
+            assertEquals(10000, default.sinkPollingInterval)
         }
         fun testFromConf(streamsConfig: StreamsSinkConfiguration, pollingInterval: String, topic: String, topicValue: String) {
             assertEquals(pollingInterval.toLong(), streamsConfig.sinkPollingInterval)

doc/asciidoc/procedures/index.adoc

@@ -58,11 +58,11 @@ Uses:
 `CALL streams.consume('my-topic', {<config>}) YIELD event RETURN event`
 
 Example:
-Imagine you have a producer that publish events like this `{"name": "Andrea", "surname": "Santurbano"}`), we can create user nodes in this way:
+Imagine you have a producer that publish events like this `{"name": "Andrea", "surname": "Santurbano"}`, we can create user nodes in this way:
 
 ```
 CALL streams.consume('my-topic', {<config>}) YIELD event
-CREATE (p:Person{firstName: event.name, lastName: event.surname})
+CREATE (p:Person{firstName: event.data.name, lastName: event.data.surname})
 ```
 
 Input Parameters:

doc/asciidoc/producer/configuration.adoc

@@ -14,13 +14,13 @@ kafka.session.timeout.ms=15000
 kafka.connection.timeout.ms=10000
 kafka.replication=1
 kafka.linger.ms=1
-kafka.transaction.id=
+kafka.transactional.id=
 
 streams.source.topic.nodes.<TOPIC_NAME>=<PATTERN>
 streams.source.topic.relationships.<TOPIC_NAME>=<PATTERN>
 streams.source.enable=<true/false, default=true>
 ----
 
-Note: To use the Kafka transactions please set `kafka.transaction.id` and `kafka.acks` properly
+Note: To use the Kafka transactions please set `kafka.transactional.id` and `kafka.acks` properly
 
-See the https://kafka.apache.org/documentation/#brokerconfigs[Apache Kafka documentation] for details on these settings.
+See the https://kafka.apache.org/documentation/#brokerconfigs[Apache Kafka documentation] for details on these settings.