Skip to content

Commit 57922d0

Browse files
conker84jexp
conker84
authored and
jexp
committed
fixes #88 (#89)
1 parent 935fa92 commit 57922d0

20 files changed

+169
-33
lines changed

Diff for: ‎assembly/assembly.xml

+21
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,21 @@
1+
<assembly
2+
xmlns="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0"
3+
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
4+
xsi:schemaLocation="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0 http://maven.apache.org/xsd/assembly-1.1.0.xsd">
5+
<id>streams-assembly-all</id>
6+
<formats>
7+
<format>jar</format>
8+
</formats>
9+
<includeBaseDirectory>false</includeBaseDirectory> <!-- strip the module prefixes -->
10+
<dependencySets>
11+
<dependencySet>
12+
<unpack>true</unpack> <!-- unpack , then repack the jars -->
13+
<useTransitiveDependencies>false</useTransitiveDependencies> <!-- do not pull in any transitive dependencies -->
14+
</dependencySet>
15+
</dependencySets>
16+
<containerDescriptorHandlers>
17+
<containerDescriptorHandler>
18+
<handlerName>metaInf-services</handlerName>
19+
</containerDescriptorHandler>
20+
</containerDescriptorHandlers>
21+
</assembly>

Diff for: ‎consumer/pom.xml

-2
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,6 @@
66

77
<groupId>org.neo4j</groupId>
88
<artifactId>neo4j-streams-consumer</artifactId>
9-
<version>3.4.7.1</version>
109
<name>Neo4j Streams - Consumer</name>
1110
<description>Neo4j Streams - Kafka Consumer</description>
1211
<packaging>jar</packaging>
@@ -17,5 +16,4 @@
1716
<version>3.4.7.1</version>
1817
</parent>
1918

20-
2119
</project>

Diff for: ‎consumer/src/main/kotlin/streams/StreamsEventSink.kt

+6
Original file line numberDiff line numberDiff line change
@@ -11,10 +11,16 @@ abstract class StreamsEventSink(private val config: Config,
1111

1212
abstract fun stop()
1313

14+
abstract fun start()
15+
1416
override fun unavailable() {
1517
stop()
1618
}
1719

20+
override fun available() {
21+
start()
22+
}
23+
1824
}
1925

2026
object StreamsEventSinkFactory {

Diff for: ‎consumer/src/main/kotlin/streams/StreamsEventSinkExtensionFactory.kt

+1-1
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,7 @@ class StreamsEventSinkExtensionFactory : KernelExtensionFactory<StreamsEventSink
2929
private val configuration = dependencies.config()
3030
private var streamsLog = log.getUserLog(StreamsEventLifecycle::class.java)
3131

32-
lateinit var eventSink: StreamsEventSink
32+
private lateinit var eventSink: StreamsEventSink
3333

3434
override fun start() {
3535
try {

Diff for: ‎consumer/src/main/kotlin/streams/StreamsSinkConfiguration.kt

+13-8
Original file line numberDiff line numberDiff line change
@@ -4,22 +4,27 @@ import org.apache.commons.lang3.StringUtils
44
import org.neo4j.kernel.configuration.Config
55

66

7-
private const val streamsConfigPrefix = "streams."
8-
private const val streamsSinkTopicPrefix = "sink.topic.cypher."
7+
private object StreamsSinkConfigurationConstants {
8+
const val STREAMS_CONFIG_PREFIX: String = "streams."
9+
const val STREAMS_SINK_TOPIC_CYPHER_PREFIX: String = "sink.topic.cypher."
10+
const val ENABLED = "sink.enabled"
11+
}
912

10-
data class StreamsSinkConfiguration(val sinkPollingInterval: Long = Long.MAX_VALUE,
13+
data class StreamsSinkConfiguration(val enabled: Boolean = true,
14+
val sinkPollingInterval: Long = Long.MAX_VALUE,
1115
val topics: Map<String, String> = emptyMap()) {
1216

1317
companion object {
1418
fun from(cfg: Config): StreamsSinkConfiguration {
1519
val default = StreamsSinkConfiguration()
1620
val config = cfg.raw
17-
.filterKeys { it.startsWith(streamsConfigPrefix) }
18-
.mapKeys { it.key.substring(streamsConfigPrefix.length) }
21+
.filterKeys { it.startsWith(StreamsSinkConfigurationConstants.STREAMS_CONFIG_PREFIX) }
22+
.mapKeys { it.key.substring(StreamsSinkConfigurationConstants.STREAMS_CONFIG_PREFIX.length) }
1923
val topics = config
20-
.filterKeys { it.startsWith(streamsSinkTopicPrefix) }
21-
.mapKeys { it.key.replace(streamsSinkTopicPrefix, StringUtils.EMPTY) }
22-
return default.copy(sinkPollingInterval = config.getOrDefault("sink.polling.interval", default.sinkPollingInterval).toString().toLong(),
24+
.filterKeys { it.startsWith(StreamsSinkConfigurationConstants.STREAMS_SINK_TOPIC_CYPHER_PREFIX) }
25+
.mapKeys { it.key.replace(StreamsSinkConfigurationConstants.STREAMS_SINK_TOPIC_CYPHER_PREFIX, StringUtils.EMPTY) }
26+
return default.copy(enabled = config.getOrDefault(StreamsSinkConfigurationConstants.ENABLED, default.enabled).toString().toBoolean(),
27+
sinkPollingInterval = config.getOrDefault("sink.polling.interval", default.sinkPollingInterval).toString().toLong(),
2328
topics = topics)
2429
}
2530
}

Diff for: ‎consumer/src/main/kotlin/streams/kafka/KafkaEventSink.kt

+2-2
Original file line numberDiff line numberDiff line change
@@ -40,15 +40,15 @@ class KafkaEventSink: StreamsEventSink {
4040
this.db = db
4141
}
4242

43-
override fun available() {
43+
override fun start() {
4444
this.streamsTopicService = StreamsTopicService(this.db, kafkaConfig.streamsSinkConfiguration)
4545
if (streamsTopicService!!.getTopics().isEmpty()) {
4646
log.info("No topic configuration found under streams.sink.topic.*, Kafka Sink will not stared")
4747
return
4848
}
4949

5050
this.queryExecution = StreamsEventSinkQueryExecution(this.streamsTopicService!!, db, log)
51-
createConsumer();
51+
createConsumer()
5252

5353
job = createJob()
5454
log.info("Kafka Sink Connector started.")

Diff for: ‎consumer/src/main/kotlin/streams/kafka/KafkaSinkConfiguration.kt

+2-2
Original file line numberDiff line numberDiff line change
@@ -61,8 +61,8 @@ data class KafkaSinkConfiguration(val zookeeperConnect: String = "localhost:2181
6161
.mapKeys { it.key.toPointCase() }
6262
props.putAll(map)
6363
props.putAll(extraProperties)
64-
props.putAll(addDeserializers())
65-
props[ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG] = true
64+
props.putAll(addDeserializers()) // Fixed deserializers
65+
props[ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG] = true // Fixed autocommit
6666
return props
6767
}
6868

Diff for: ‎distribution/pom.xml

+59
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,59 @@
1+
<?xml version="1.0" encoding="UTF-8"?>
2+
<project xmlns="http://maven.apache.org/POM/4.0.0"
3+
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
4+
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
5+
<modelVersion>4.0.0</modelVersion>
6+
7+
<groupId>org.neo4j</groupId>
8+
<artifactId>neo4j-streams-distribution</artifactId>
9+
<name>Neo4j Streams - Distribution</name>
10+
<description>Neo4j Streams - Distribution Package</description>
11+
<packaging>pom</packaging>
12+
13+
<parent>
14+
<groupId>org.neo4j</groupId>
15+
<artifactId>neo4j-streams-parent</artifactId>
16+
<version>3.4.7.1</version>
17+
</parent>
18+
19+
<!-- Include here all the required dependencies of the final package -->
20+
<dependencies>
21+
<dependency>
22+
<groupId>org.neo4j</groupId>
23+
<artifactId>neo4j-streams-producer</artifactId>
24+
<version>${project.version}</version>
25+
</dependency>
26+
<dependency>
27+
<groupId>org.neo4j</groupId>
28+
<artifactId>neo4j-streams-consumer</artifactId>
29+
<version>${project.version}</version>
30+
</dependency>
31+
</dependencies>
32+
33+
<build>
34+
<plugins>
35+
<plugin>
36+
<artifactId>maven-assembly-plugin</artifactId>
37+
<version>2.4</version>
38+
<executions>
39+
<execution>
40+
<id>streams-assembly</id>
41+
<phase>package</phase> <!-- create assembly in package phase (invoke 'single' goal on assemby plugin)-->
42+
<goals>
43+
<goal>single</goal>
44+
</goals>
45+
<configuration>
46+
<descriptors>
47+
<descriptor>assembly/assembly.xml</descriptor>
48+
</descriptors>
49+
<appendAssemblyId>false</appendAssemblyId>
50+
<finalName>neo4j-streams-${project.version}</finalName>
51+
<appendAssemblyId>false</appendAssemblyId>
52+
<outputDirectory>../target</outputDirectory>
53+
</configuration>
54+
</execution>
55+
</executions>
56+
</plugin>
57+
</plugins>
58+
</build>
59+
</project>

Diff for: ‎doc/asciidoc/consumer/configuration.adoc

+2
Original file line numberDiff line numberDiff line change
@@ -6,8 +6,10 @@ kafka.zookeeper.connect=localhost:2181
66
kafka.bootstrap.servers=localhost:9092
77
kafka.auto.offset.reset=earliest
88
kafka.group.id=neo4j
9+
910
streams.sink.polling.interval=<The time, in milliseconds, spent waiting in poll if data is not available in the buffer. default=Long.MAX_VALUE>
1011
streams.sink.topic.cypher.<TOPIC_NAME>=<CYPHER_QUERY>
12+
streams.sink.enable=<true/false, default=true>
1113
----
1214

1315
See the https://kafka.apache.org/documentation/#brokerconfigs[Apache Kafka documentation] for details on these settings.

Diff for: ‎doc/asciidoc/producer/configuration.adoc

+2
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,8 @@ kafka.transaction.id=
1818
1919
streams.source.topic.nodes.<TOPIC_NAME>=<PATTERN>
2020
streams.source.topic.relationships.<TOPIC_NAME>=<PATTERN>
21+
streams.source.enable=<true/false, default=true>
22+
streams.procedures.enable=<true/false, default=true>
2123
----
2224

2325
Note: To use the Kafka transactions please set `kafka.transaction.id` and `kafka.acks` properly

Diff for: ‎performance/docker-compose.yml

+2
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@ services:
2121
NEO4J_dbms_memory_heap_max__size: 8G
2222
# NEO4J_dbms_logs_debug_level: DEBUG
2323
NEO4J_kafka_batch_size: 16384
24+
NEO4J_streams_sink_enabled: "false"
2425

2526
neo4j-consumer:
2627
image: neo4j:3.4
@@ -40,6 +41,7 @@ services:
4041
NEO4J_AUTH: neo4j/consumer
4142
NEO4J_dbms_memory_heap_max__size: 2G
4243
NEO4J_kafka_max_poll_records: 16384
44+
NEO4J_streams_source_enabled: "false"
4345
NEO4J_streams_sink_topic_cypher_neo4j: "WITH event.payload AS payload, event.meta AS meta CALL apoc.do.case( [
4446
payload.type = 'node' AND meta.operation = 'created', \
4547
'CREATE (x:Performance {received_time: apoc.date.currentTimestamp()}) SET x+=props RETURN count(x)']

Diff for: ‎pom.xml

+2-1
Original file line numberDiff line numberDiff line change
@@ -23,14 +23,15 @@
2323
<module>producer</module>
2424
<module>consumer</module>
2525
<module>doc</module>
26+
<module>distribution</module>
2627
</modules>
2728

2829
<properties>
2930
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
3031
<java.version>1.8</java.version>
3132
<kotlin.version>1.3.0</kotlin.version>
3233
<kotlin.coroutines.version>1.0.0</kotlin.coroutines.version>
33-
<neo4j.version>3.4.7</neo4j.version>
34+
<neo4j.version>3.4.10</neo4j.version>
3435
<kafka.version>1.0.1</kafka.version>
3536
<kotlin.compiler.incremental>true</kotlin.compiler.incremental>
3637
</properties>

Diff for: ‎producer/pom.xml

-2
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,6 @@
66

77
<groupId>org.neo4j</groupId>
88
<artifactId>neo4j-streams-producer</artifactId>
9-
<version>3.4.7.1</version>
109
<name>Neo4j Streams - Producer</name>
1110
<description>Neo4j Streams - Kafka Producer</description>
1211
<packaging>jar</packaging>
@@ -17,5 +16,4 @@
1716
<version>3.4.7.1</version>
1817
</parent>
1918

20-
2119
</project>

Diff for: ‎producer/src/main/kotlin/streams/Extensions.kt

+4
Original file line numberDiff line numberDiff line change
@@ -20,4 +20,8 @@ fun Relationship.toMap(): Map<String, Any?> {
2020

2121
fun Node.labelNames() : List<String> {
2222
return this.labels.map { it.name() }
23+
}
24+
25+
fun String.toPointCase(): String {
26+
return this.split("(?<=[a-z])(?=[A-Z])".toRegex()).joinToString(separator = ".").toLowerCase()
2327
}

Diff for: ‎producer/src/main/kotlin/streams/StreamsEventRouterConfiguration.kt

+4-1
Original file line numberDiff line numberDiff line change
@@ -12,10 +12,12 @@ private fun <T> filterMap(config: Map<String, String>, routingPrefix: String): L
1212
private object StreamsRoutingConfigurationConstants {
1313
const val NODE_ROUTING_KEY_PREFIX: String = "streams.source.topic.nodes."
1414
const val REL_ROUTING_KEY_PREFIX: String = "streams.source.topic.relationships."
15-
const val ENABLED = "streams.produce.enabled"
15+
const val ENABLED = "streams.source.enabled"
16+
const val PROCEDURES_ENABLED = "streams.procedures.enabled"
1617
}
1718

1819
data class StreamsEventRouterConfiguration(val enabled: Boolean = true,
20+
val proceduresEnabled: Boolean = true,
1921
val nodeRouting: List<NodeRoutingConfiguration> = listOf(NodeRoutingConfiguration()),
2022
val relRouting: List<RelationshipRoutingConfiguration> = listOf(RelationshipRoutingConfiguration())) {
2123
companion object {
@@ -28,6 +30,7 @@ data class StreamsEventRouterConfiguration(val enabled: Boolean = true,
2830

2931
val default = StreamsEventRouterConfiguration()
3032
return default.copy(enabled = config.getOrDefault(StreamsRoutingConfigurationConstants.ENABLED, default.enabled).toString().toBoolean(),
33+
proceduresEnabled = config.getOrDefault(StreamsRoutingConfigurationConstants.PROCEDURES_ENABLED, default.proceduresEnabled).toString().toBoolean(),
3134
nodeRouting = if (nodeRouting.isEmpty()) default.nodeRouting else nodeRouting,
3235
relRouting = if (relRouting.isEmpty()) default.relRouting else relRouting)
3336
}

Diff for: ‎producer/src/main/kotlin/streams/StreamsExtensionFactory.kt

+7-3
Original file line numberDiff line numberDiff line change
@@ -45,12 +45,16 @@ class StreamsEventRouterLifecycle(val db: GraphDatabaseAPI, val streamHandler: S
4545
}
4646

4747
fun registerTransactionEventHandler() {
48-
txHandler = StreamsTransactionEventHandler(streamHandler, streamsEventRouterConfiguration)
49-
db.registerTransactionEventHandler(txHandler)
48+
if (streamsEventRouterConfiguration.enabled) {
49+
txHandler = StreamsTransactionEventHandler(streamHandler, streamsEventRouterConfiguration)
50+
db.registerTransactionEventHandler(txHandler)
51+
}
5052
}
5153

5254
fun unregisterTransactionEventHandler() {
53-
db.unregisterTransactionEventHandler(txHandler)
55+
if (streamsEventRouterConfiguration.enabled) {
56+
db.unregisterTransactionEventHandler(txHandler)
57+
}
5458
}
5559

5660
override fun stop() {

Diff for: ‎producer/src/main/kotlin/streams/kafka/KafkaConfiguration.kt

+22-7
Original file line numberDiff line numberDiff line change
@@ -5,8 +5,10 @@ import org.apache.kafka.clients.producer.ProducerConfig
55
import org.apache.kafka.common.serialization.ByteArraySerializer
66
import org.apache.kafka.common.serialization.LongSerializer
77
import org.apache.kafka.common.serialization.StringSerializer
8+
import org.codehaus.jackson.map.ObjectMapper
89
import streams.getInt
910
import streams.serialization.JacksonUtil
11+
import streams.toPointCase
1012
import java.util.*
1113

1214
private val configPrefix = "kafka."
@@ -23,12 +25,23 @@ data class KafkaConfiguration(val zookeeperConnect: String = "localhost:2181",
2325
val connectionTimeoutMs: Int = 10 * 1000,
2426
val replication: Int = 1,
2527
val transactionalId: String = StringUtils.EMPTY,
26-
val lingerMs: Int = 1){
28+
val lingerMs: Int = 1,
29+
val extraProperties: Map<String, String> = emptyMap()) {
30+
31+
private fun asMap(): Map<String, Any?> {
32+
return ObjectMapper().convertValue(this, Map::class.java)
33+
.mapKeys { it.key.toString() }
34+
}
35+
2736
companion object {
2837
fun from(cfg: Map<String, String>) : KafkaConfiguration {
2938
val config = cfg.filterKeys { it.startsWith(configPrefix) }.mapKeys { it.key.substring(configPrefix.length) }
3039

3140
val default = KafkaConfiguration()
41+
42+
val keys = default.asMap().keys.map { it.toPointCase() }
43+
val extraProperties = config.filterKeys { !keys.contains(it) }
44+
3245
return default.copy(zookeeperConnect = config.getOrDefault("zookeeper.connect",default.zookeeperConnect),
3346
bootstrapServers = config.getOrDefault("bootstrap.servers", default.bootstrapServers),
3447
acks = config.getOrDefault("acks", default.acks),
@@ -41,24 +54,26 @@ data class KafkaConfiguration(val zookeeperConnect: String = "localhost:2181",
4154
connectionTimeoutMs = config.getInt("connection.timeout.ms", default.connectionTimeoutMs),
4255
replication = config.getInt("replication", default.replication),
4356
transactionalId = config.getOrDefault("transactional.id", default.transactionalId),
44-
lingerMs = config.getInt("linger.ms", default.lingerMs)
57+
lingerMs = config.getInt("linger.ms", default.lingerMs),
58+
extraProperties = extraProperties // for what we don't provide a default configuration
4559
)
4660
}
4761
}
4862

4963
fun asProperties(): Properties {
5064
val props = Properties()
51-
val map = JacksonUtil.getMapper().convertValue(this, Map::class.java)
52-
.mapKeys { it.key.toString().split("(?<=[a-z])(?=[A-Z])".toRegex()).joinToString(separator = ".").toLowerCase() }
65+
val map = this.asMap()
5366
.filter {
54-
if (it.key == "transactional.id") {
67+
if (it.key == "transactionalId") {
5568
it.value != StringUtils.EMPTY
5669
} else {
57-
it.key != "node.routing" && it.key != "rel.routing"
70+
true
5871
}
5972
}
73+
.mapKeys { it.key.toPointCase() }
6074
props.putAll(map)
61-
props.putAll(addSerializers())
75+
props.putAll(extraProperties)
76+
props.putAll(addSerializers()) // Fixed serializers
6277
return props
6378
}
6479

0 commit comments

Comments
 (0)