Issue with kafka connet
Avi Zrachya
We moved to the paid instances of neo4j
Ever since we did we keep getting these errors from the neo4j connector sink
It would be ok if after this error it would just reconnect, but i's not... this error continues until we restart the connector.
Please help with it, otherwise you force us to move to our own instances.
connect | at org.neo4j.driver.internal.util.Futures.blockingGet(Futures.java:111)
connect | at streams.kafka.connect.sink.Neo4jSinkService$write$1$1$1.invoke$lambda-0(Neo4jSinkService.kt:50)
connect | at org.neo4j.driver.internal.InternalSession.lambda$transaction$4(InternalSession.java:137)
connect | at org.neo4j.driver.internal.retry.ExponentialBackoffRetryLogic.retry(ExponentialBackoffRetryLogic.java:106)
connect | at org.neo4j.driver.internal.InternalSession.transaction(InternalSession.java:134)
connect | at org.neo4j.driver.internal.InternalSession.writeTransaction(InternalSession.java:113)
connect | at streams.kafka.connect.sink.Neo4jSinkService$write$1$1$1.invoke(Neo4jSinkService.kt:49)
connect | at streams.kafka.connect.sink.Neo4jSinkService$write$1$1$1.invoke(Neo4jSinkService.kt:46)
connect | at streams.utils.CoroutineUtilsKt.retryForException(CoroutineUtils.kt:7)
connect | at streams.utils.CoroutineUtilsKt.retryForException(CoroutineUtils.kt:12)
connect | at streams.utils.CoroutineUtilsKt.retryForException(CoroutineUtils.kt:12)
connect | at streams.utils.CoroutineUtilsKt.retryForException(CoroutineUtils.kt:12)
connect | at streams.utils.CoroutineUtilsKt.retryForException(CoroutineUtils.kt:12)
connect | at streams.utils.CoroutineUtilsKt.retryForException(CoroutineUtils.kt:12)
connect | at streams.kafka.connect.sink.Neo4jSinkService$write$1$1.invokeSuspend(Neo4jSinkService.kt:46)
connect | at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
connect | at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:106)
connect | at kotlinx.coroutines.EventLoopImplBase.processNextEvent(EventLoop.common.kt:284)
connect | at kotlinx.coroutines.BlockingCoroutine.joinBlocking(Builders.kt:85)
connect | at kotlinx.coroutines.BuildersKt__BuildersKt.runBlocking(Builders.kt:59)
connect | at kotlinx.coroutines.BuildersKt.runBlocking(Unknown Source)
connect | at kotlinx.coroutines.BuildersKt__BuildersKt.runBlocking$default(Builders.kt:38)
connect | at kotlinx.coroutines.BuildersKt.runBlocking$default(Unknown Source)
connect | at streams.kafka.connect.sink.Neo4jSinkService.write(Neo4jSinkService.kt:45)
connect | at streams.service.StreamsSinkService.writeWithStrategy(StreamsSinkService.kt:32)
connect | at streams.service.StreamsSinkService.writeForTopic(StreamsSinkService.kt:40)
connect | at streams.kafka.connect.sink.Neo4jSinkService$writeDataAsync$1$jobs$1$1$1.invokeSuspend(Neo4jSinkService.kt:84)
connect | at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
connect | at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:106)
connect | at kotlinx.coroutines.internal.LimitedDispatcher.run(LimitedDispatcher.kt:42)
connect | at kotlinx.coroutines.scheduling.TaskImpl.run(Tasks.kt:95)
connect | at kotlinx.coroutines.scheduling.CoroutineScheduler.runSafely(CoroutineScheduler.kt:570)
connect | at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.executeTask(CoroutineScheduler.kt:750)
connect | at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.runWorker(CoroutineScheduler.kt:677)
connect | at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.run(CoroutineScheduler.kt:664)
connect | Suppressed: org.neo4j.driver.internal.util.ErrorUtil$InternalExceptionCause
connect | at org.neo4j.driver.internal.util.ErrorUtil.newNeo4jError(ErrorUtil.java:76)
connect | at org.neo4j.driver.internal.async.inbound.InboundMessageDispatcher.handleFailureMessage(InboundMessageDispatcher.java:107)
connect | at org.neo4j.driver.internal.messaging.common.CommonMessageReader.unpackFailureMessage(CommonMessageReader.java:75)
connect | at org.neo4j.driver.internal.messaging.common.CommonMessageReader.read(CommonMessageReader.java:53)
connect | at org.neo4j.driver.internal.async.inbound.InboundMessageHandler.channelRead0(InboundMessageHandler.java:81)
connect | at org.neo4j.driver.internal.async.inbound.InboundMessageHandler.channelRead0(InboundMessageHandler.java:37)
connect | at org.neo4j.driver.internal.shaded.io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:99)
connect | at org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
connect | at org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
connect | at org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
connect | at org.neo4j.driver.internal.shaded.io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:327)
connect | at org.neo4j.driver.internal.shaded.io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:299)
connect | at org.neo4j.driver.internal.async.inbound.MessageDecoder.channelRead(MessageDecoder.java:42)
connect | at org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
connect | at org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
connect | at org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
connect | at org.neo4j.driver.internal.shaded.io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:327)
connect | at org.neo4j.driver.internal.shaded.io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:314)
connect | at org.neo4j.driver.internal.shaded.io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:435)
connect | at org.neo4j.driver.internal.shaded.io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:279)
connect | at org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
connect | at org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
connect | at org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
connect | at org.neo4j.driver.internal.shaded.io.netty.handler.ssl.SslHandler.unwrap(SslHandler.java:1373)
connect | at org.neo4j.driver.internal.shaded.io.netty.handler.ssl.SslHandler.decodeJdkCompatible(SslHandler.java:1236)
connect | at org.neo4j.driver.internal.shaded.io.netty.handler.ssl.SslHandler.decode(SslHandler.java:1285)
connect | at org.neo4j.driver.internal.shaded.io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:510)
connect | at org.neo4j.driver.internal.shaded.io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:449)
connect | at org.neo4j.driver.internal.shaded.io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:279)
connect | at org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
connect | at org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
connect | at org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
connect | at org.neo4j.driver.internal.shaded.io.netty.handler.timeout.IdleStateHandler.channelRead(IdleStateHandler.java:286)
connect | at org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
connect | at org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
connect | at org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
connect | at org.neo4j.driver.internal.shaded.io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
connect | at org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
connect | at org.neo4j.driver.internal.shaded.io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
connect | at org.neo4j.driver.internal.shaded.io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
connect | at org.neo4j.driver.internal.shaded.io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
connect | at org.neo4j.driver.internal.shaded.io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:722)
connect | at org.neo4j.driver.internal.shaded.io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:658)
connect | at org.neo4j.driver.internal.shaded.io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:584)
connect | at org.neo4j.driver.internal.shaded.io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:496)
connect | at org.neo4j.driver.internal.shaded.io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
connect | at org.neo4j.driver.internal.shaded.io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
connect | at org.neo4j.driver.internal.shaded.io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
connect | at java.base/java.lang.Thread.run(Thread.java:829)
connect | [2023-05-24 07:15:37,195] ERROR ErrorData(originalTopic=sink.neo4j.hierarchy.v0, timestamp=1684910239112, partition=0, offset=1216743, exception=org.neo4j.driver.exceptions.ClientException: No write operations are allowed directly on this database. Writes must pass through the leader. The role of this server is: FOLLOWER, key=null, value={op=merge, rel_type=MEMBER, from={op=merge, ids={uuid=e94c6760-00d8-45e5-85a5-caabb915ace9}, labels=[employees]}, to={op=merge, ids={uuid=0326d36a-bef8-4fe4-a5d2-35d5557b75fc}, labels=[departments]}, , executingClass=class streams.kafka.connect.sink.Neo4jSinkTask) (streams.kafka.connect.sink.Neo4jSinkTask)
connect | org.neo4j.driver.exceptions.ClientException: No write operations are allowed directly on this database. Writes must pass through the leader. The role of this server is: FOLLOWER