bug fix pre release (#290)

Author: conker84

common/src/main/kotlin/streams/extensions/GraphDatabaseServerExtensions.kt

@@ -8,11 +8,8 @@ fun GraphDatabaseService.execute(cypher: String) = this.execute(cypher, emptyMap
 fun GraphDatabaseService.execute(cypher: String, params: Map<String, Any>) = this.executeTransactionally(cypher, params)
 
 fun <T> GraphDatabaseService.execute(cypher: String, lambda: ((Result) -> T)) = this.execute(cypher, emptyMap(), lambda)
-fun <T> GraphDatabaseService.execute(cypher: String, params: Map<String, Any>, lambda: ((Result) -> T)) = this.beginTx().use {
-    val result = it.execute(cypher, params)
-    val ret = lambda(result)
-    it.commit()
-    ret
-}
+fun <T> GraphDatabaseService.execute(cypher: String,
+                                     params: Map<String, Any>,
+                                     lambda: ((Result) -> T)) = this.executeTransactionally(cypher, params, lambda)
 
 fun GraphDatabaseService.isSystemDb() = this.databaseName() == Neo4jUtils.SYSTEM_DATABASE_NAME

consumer/src/test/kotlin/integrations/kafka/KafkaEventSinkNoTopicAutocreationIT.kt

@@ -52,7 +52,7 @@ class KafkaEventSinkNoTopicAutoCreationIT {
                 exists = true
             }, IllegalStateException::class.java)
             Assume.assumeTrue("Kafka container has to exist", exists)
-            Assume.assumeTrue("Kafka must be running", kafka.isRunning)
+            Assume.assumeTrue("Kafka must be running", ::kafka.isInitialized && kafka.isRunning)
         }
 
         @AfterClass

consumer/src/test/kotlin/integrations/kafka/KafkaEventSinkSuiteIT.kt

@@ -52,7 +52,7 @@ class KafkaEventSinkSuiteIT {
                 schemaRegistry.start()
                 isRunning = true
             }, IllegalStateException::class.java)
-            assumeTrue("Kafka must be running", kafka.isRunning)
+            assumeTrue("Kafka must be running", ::kafka.isInitialized && kafka.isRunning)
             assumeTrue("Schema Registry must be running", schemaRegistry.isRunning)
             assumeTrue("isRunning must be true", isRunning)
         }

consumer/src/test/kotlin/integrations/kafka/KafkaStreamsSinkProceduresTSE.kt

@@ -183,6 +183,7 @@ class KafkaStreamsSinkProceduresTSE : KafkaEventSinkBaseTSE() {
             assertTrue { searchResultMap.containsKey("count") }
             assertEquals(1L, searchResultMap["count"])
         }
+        Unit
     }
 
     @Test

pom.xml

@@ -50,7 +50,7 @@
         <java.version>1.8</java.version>
         <kotlin.version>1.3.61</kotlin.version>
         <kotlin.coroutines.version>1.3.3</kotlin.coroutines.version>
-        <neo4j.version>4.0.0</neo4j.version>
+        <neo4j.version>4.0.1</neo4j.version>
         <kafka.version>2.3.0</kafka.version>
         <jackson.version>2.9.7</jackson.version>
         <kotlin.compiler.incremental>true</kotlin.compiler.incremental>

producer/src/main/kotlin/streams/StreamsExtensionFactory.kt

@@ -86,10 +86,12 @@ class StreamsEventRouterLifecycle(private val db: GraphDatabaseAPI,
     }
 
     private fun unregisterTransactionEventHandler() {
-        if (streamsEventRouterConfiguration.enabled) {
-            StreamsUtils.ignoreExceptions({ streamsConstraintsService.close() }, UninitializedPropertyAccessException::class.java)
-            StreamsUtils.ignoreExceptions({ databaseManagementService.unregisterTransactionEventListener(db.databaseName(), txHandler) }, UninitializedPropertyAccessException::class.java)
-        }
+        StreamsUtils.ignoreExceptions({
+            if (streamsEventRouterConfiguration.enabled) {
+                StreamsUtils.ignoreExceptions({ streamsConstraintsService.close() }, UninitializedPropertyAccessException::class.java)
+                StreamsUtils.ignoreExceptions({ databaseManagementService.unregisterTransactionEventListener(db.databaseName(), txHandler) }, UninitializedPropertyAccessException::class.java)
+            }
+        }, UninitializedPropertyAccessException::class.java)
     }
 
     override fun stop() {

producer/src/test/kotlin/streams/integrations/KafkaEventRouterBaseTSE.kt

@@ -9,7 +9,6 @@ import org.neo4j.test.rule.DbmsRule
 import org.neo4j.test.rule.ImpermanentDbmsRule
 import streams.KafkaTestUtils
 import streams.setConfig
-import streams.shutdownSilently
 
 open class KafkaEventRouterBaseTSE { // TSE (Test Suit Element)
 
@@ -47,7 +46,7 @@ open class KafkaEventRouterBaseTSE { // TSE (Test Suit Element)
 
     @After
     fun tearDown() {
-        db.shutdownSilently()
+        db.shutdown()
         kafkaConsumer.close()
     }
 }

producer/src/test/kotlin/streams/integrations/KafkaEventRouterNoTopicAutocreationIT.kt

@@ -15,7 +15,6 @@ import streams.extensions.execute
 import streams.kafka.KafkaConfiguration
 import streams.KafkaTestUtils
 import streams.setConfig
-import streams.shutdownSilently
 import streams.start
 import streams.utils.StreamsUtils
 import kotlin.test.assertEquals
@@ -50,7 +49,7 @@ class KafkaEventRouterNoTopicAutocreationIT {
                 exists = true
             }, IllegalStateException::class.java)
             Assume.assumeTrue("Kafka container has to exist", exists)
-            Assume.assumeTrue("Kafka must be running", kafka.isRunning)
+            Assume.assumeTrue("Kafka must be running", ::kafka.isInitialized && kafka.isRunning)
 
             val client = AdminClient.create(mapOf("bootstrap.servers" to "localhost:" + kafka.firstMappedPort))
             val topicsToCreate = listOf("person")
@@ -81,7 +80,7 @@ class KafkaEventRouterNoTopicAutocreationIT {
             it.columnAs<Long>("count").next()
         }
         assertEquals(0L, count)
-        db.shutdownSilently()
+        db.shutdown()
     }
 
     @Test
@@ -116,6 +115,7 @@ class KafkaEventRouterNoTopicAutocreationIT {
         // then
         val count = db.execute("MATCH (n) RETURN COUNT(n) AS count") { it.columnAs<Long>("count").next() }
         assertEquals(2L, count)
+        db.shutdown()
     }
 
 }

producer/src/test/kotlin/streams/integrations/KafkaEventRouterSuiteIT.kt

@@ -48,7 +48,7 @@ class KafkaEventRouterSuiteIT {
                 exists = true
             }, IllegalStateException::class.java)
             Assume.assumeTrue("Kafka container has to exist", exists)
-            Assume.assumeTrue("Kafka must be running", kafka.isRunning)
+            Assume.assumeTrue("Kafka must be running", ::kafka.isInitialized && kafka.isRunning)
         }
 
         @AfterClass @JvmStatic

producer/src/test/kotlin/streams/integrations/StreamsTransactionEventHandlerIT.kt

@@ -12,7 +12,6 @@ import streams.events.RelationshipPayload
 import streams.extensions.execute
 import streams.mocks.MockStreamsEventRouter
 import streams.setConfig
-import streams.shutdownSilently
 import streams.start
 import kotlin.test.assertEquals
 import kotlin.test.assertNotNull
@@ -31,10 +30,10 @@ class StreamsTransactionEventHandlerIT {
 
     @After
     fun tearDown() {
-        db.shutdownSilently()
+        db.shutdown()
     }
 
-    @Test fun testNodes(){
+    @Test fun testNodes() {
         db.execute("CREATE (:Person {name:'Omar', age: 30}), (:Person {name:'Andrea', age: 31})")
 
         assertEquals(2,MockStreamsEventRouter.events.size)