Strange behavior of the streams.publish procedure inside apoc.periodic.repeat procedure

[doctor3030]

Please note that GitHub issues are only meant for bug reports/feature requests. If you have questions on how to use Streams, please ask on StackOverflow instead of creating an issue here.

Assume a background process:
CALL apoc.periodic.repeat("process_name",
"MATCH (n:Label)
WITH n.prp1 AS val1, n.prop2 AS val2
CALL streams.publish("kafka_topic", {val1: val1, val2: val2})
RETURN null", 20)

When created, this process does not send anything to kafka.
When the streams.publish is replaced with a dummy statement like counter = counter + 1, it works.
When the streams.publish is called manualy outside the apoc.periodic.repeat it also works (messages are published).

I was able to solve this issue by calling streams.publish from a dummy apoc.do.when procedure like this:
CALL apoc.periodic.repeat("process_name",
"MATCH (n:Label)
WITH n.prp1 AS val1, n.prop2 AS val2
CALL apoc.do.when(true,
'CALL streams.publish(\"kafka_topic\", {val1: val1, val2: val2}) RETURN null',
'RETURN null', {val1: val1, val2: val2}) YIELD value
RETURN null", 20)

Maybe I missing something but this behavior is not documented anywhere and I was able to find a solution only because I already had a similar periodic.repeat process that already had a streams.publish inside apoc.do.when and worked perfectly.

Versions

• OS: ubuntu 20.04
• Neo4j: desktop enterprice 4.2.5
• Neo4j-APOC: 4.2.0.5
• Neo4j-Streams: 4.0.8

[mroiter-larus]

Hi @doctor3030,

sorry for the late reply. Have you already solved your issue?
If not, could you please share your neo4j.conf file, and in particular the Streams related part?

Regards,

Mauro

[doctor3030]

Well Im still using dummy apoc.do.when as I described above.

Here is my neo4j.conf
neo4j.conf.txt

Streams.conf
kafka.bootstrap.servers=10.0.0.120:19092,10.0.0.224:19092,10.0.0.114:19092
kafka.auto.offset.reset=earliest
kafka.group.id=neo4j
kafka.message.max.bytes=1048576000
kafka.enable.auto.commit=false
kafka.streams.async.commit=false
kafka.key.deserializer=org.apache.kafka.common.serialization.ByteArrayDeserializer
kafka.value.deserializer=org.apache.kafka.common.serialization.ByteArrayDeserializer
kafka.session.timeout.ms=25000

streams.source.enabled=true
streams.sink.enabled=true
streams.sink.enabled.to.rehoboam=true
streams.source.enabled.from.rehoboam=true
streams.procedures.enabled.from.rehoboam=true
streams.source.schema.polling.interval=300000

Thanks,
Dmitry

[mroiter-larus]

Hi @doctor3030,

the issue is caused by how apoc works. I mean apoc.periodic.repeate open a new transaction and this is why it doesn't works together with Streams. If you need to use it, the workaround you already found is currently the only solution.

Regards,

Mauro