Neo4j and Kafka containers seem to be unable to communicate with each other

[jameswylyreasint]

EDIT: Hardcoded connection string 'kafka:9092' with an explicit network alias of 'kafka' on the KafkaContainer ended up being the answer. Slight variations on the format fail and it is not returned from any internal methods in a way which can be used for connection strings.

Hello I have been working on converting some our existing neo4j plugin tests from the legacy neo4j test-harness over to testcontainers. For most uses cases this is working as expected, but in cases where we try to connect to testcontainers kafka from our neo4j testcontainer we're seeing some unexpected results when using docker networking and the neo4j streams kafka plugin.

General overview :

//Create shared network
Network sharedNetwork = Network.newNetwork();

//Create kafka container
KafkaContainer kafkaContainer = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:6.2.1")).withNetwork(sharedNetwork);

//Start kafka container FIRST so we can read dynamic values
kafkaContainer.start();

//Create neo4j container with streams configuration for kafkaContainer dynamic ports
Neo4jContainer container = new Neo4jContainer<>(DockerImageName.parse("neo4j:4.4.11-enterprise))
  .withNetwork(sharedNetwork)
  .withNeo4jConfig("kafka.bootstrap.servers", kafkaContainer.getNetworkAliases().get(0)+":9093")//Have tried any number of combinations here
  ... (more configs : memory, add streams plugin,etc)
  
  container.start();

Both containers start as expected, but no matter the value I put for kafka.bootstrap.servers I am unable to get the neo4j container to communicate with the kafka container.

Typical error message:

WARN org.apache.kafka.clients.NetworkClient - [AdminClient clientId=adminclient-1] Connection to node 1 (localhost/127.0.0.1:55291) could not be established. Broker may not be available.

The address here is notable... for most formatted kafka.bootstrap.servers values it resolves to 'localhost' plus the dynamic TCP port assigned from testcontainers pointing to the KafkaContainers.KAFKA_PORT value (9093 on version 1.17.4). The raw value assigned in the code is present within the streams.conf file as expected on the neo4j container, but it somehow being converted at runtime in an unexpected way. The only time this conversion does NOT occur is if I use the format from kafakContainer.getBootstrapServers() which outputs a format such as 'PLAINTEXT://{ALIAS_NAME}:{DYNAMIC_PORT}' and then the errors show those values ({ALIAS_NAME}/{CONTAINER_IP}:{DYNAMIC_PORT} .

It's not clear to my why these values are being converted (or which piece of code is responsible for this). With our docker compose configurations we can simply use 'kafka.bootstrap.servers={ALIAS}:{PORT}'.

I've tried assigned explicit aliases and using hardcoded configuraions, mounting configuration files directly, but nearly every solution results in the 'localhost' values being used at runtime.

Is there something I'm misconfiguring here, or is there a potential issue with trying to get these two containers to interact in this way through testcontainers?

[eddumelendez]

@jameswylyreasint can you share a minimal sample reproducer, please? I have

@Test
void test() {
    Network network = Network.newNetwork();
    try (KafkaContainer kafka = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:6.2.1"))
            .withNetwork(network)
            .withNetworkAliases("kafka");
         Neo4jContainer<?> neo4j = new Neo4jContainer<>(DockerImageName.parse("neo4j:4.4.11-enterprise"))
                 .withNetwork(network)
                 .withEnv("NEO4J_kafka_bootstrap_servers", "kafka:9093")
                 .withEnterpriseEdition()
                 .withLabsPlugins(Neo4jLabsPlugin.STREAMS)
                 .dependsOn(kafka)) {
         Startables.deepStart(kafka, neo4j).join();
    }
}

and I'm not able to see the WARN. I follow this but looks like there are more containers involved, want to make sure I'm not missing something from your report.

[eddumelendez]

When using withLabsPlugins(Neo4jLabsPlugin.STREAMS, Neo4jLabsPlugin.APOC), those are installed and as you can see in logs neo4j-streams-4.1.2.jar and apoc-4.4.0.10-all.jar are used.

STDOUT: Changed password for user 'neo4j'. IMPORTANT: this change will only take effect if performed before the database is started for the first time.
STDOUT: Fetching versions.json for Plugin 'streams' from https://neo4j-contrib.github.io/neo4j-streams/versions.json
STDOUT: Installing Plugin 'streams' from https://github.com/neo4j-contrib/neo4j-streams/releases/download/4.1.2/neo4j-streams-4.1.2.jar to /var/lib/neo4j/plugins/streams.jar 
STDOUT: Applying default values for plugin streams to neo4j.conf
STDOUT: Fetching versions.json for Plugin 'apoc' from https://neo4j-contrib.github.io/neo4j-apoc-procedures/versions.json
STDOUT: Installing Plugin 'apoc' from https://github.com/neo4j-contrib/neo4j-apoc-procedures/releases/download/4.4.0.10/apoc-4.4.0.10-all.jar to /var/lib/neo4j/plugins/apoc.jar 
STDOUT: Applying default values for plugin apoc to neo4j.conf
STDOUT: 2022-11-18 18:41:03.585+0000 WARN  Unrecognized setting. No declared setting with name: kafka.bootstrap.servers
STDOUT: 2022-11-18 18:41:03.605+0000 INFO  Starting...
STDOUT: 2022-11-18 18:41:04.057+0000 INFO  This instance is ServerId{8512fa71} (8512fa71-13f6-44ed-bc08-17441a7a0417)
STDOUT: 2022-11-18 18:41:05.855+0000 INFO  ======== Neo4j 4.4.11 ========
STDOUT: 2022-11-18 18:41:14.659+0000 INFO  [neo4j/34654bc9] Configuration has found the following environment variables: {kafka.bootstrap.servers=kafka:9093}
STDOUT: 2022-11-18 18:41:14.663+0000 INFO  [neo4j/34654bc9] Adding listener for event type CONFIGURATION_INITIALIZED
STDOUT: 2022-11-18 18:41:14.718+0000 INFO  [neo4j/34654bc9] StreamsConfig started
STDOUT: 2022-11-18 18:41:14.716+0000 INFO  [neo4j/34654bc9] Starting the connector lifecycle listener
STDOUT: 2022-11-18 18:41:14.666+0000 INFO  [neo4j/34654bc9] Adding listener for event type CONFIGURATION_INITIALIZED
STDOUT: 2022-11-18 18:41:14.661+0000 INFO  [neo4j/34654bc9] Saving the configuration back to the file
STDOUT: 2022-11-18 18:41:16.484+0000 INFO  Called db.clearQueryCaches(): Query caches successfully cleared of 1 queries.
STDOUT: 2022-11-18 18:41:16.855+0000 INFO  Sending metrics to CSV file at /var/lib/neo4j/metrics
STDOUT: 2022-11-18 18:41:16.882+0000 INFO  Bolt enabled on 0.0.0.0:7687.

I also tried mounting plugins using apoc-4.4.0.11-all.jar and neo4j-streams-4.1.2.jar. I have increased the sleep up to 100 sec and nothing :/

11:02:26.481 INFO  🐳 [neo4j:4.4.11-enterprise] - Container neo4j:4.4.11-enterprise started in PT25.695341S
2022-11-18 19:02:32.962+0000 INFO  [neo4j/d68a0fec] [Sink] An event change is detected CONFIGURATION_INITIALIZED
2022-11-18 19:02:34.896+0000 INFO  [neo4j/d68a0fec] [Sink] Registering the Streams Sink procedures
2022-11-18 19:02:34.897+0000 INFO  [neo4j/d68a0fec] [Source] An event change is detected CONFIGURATION_INITIALIZED
2022-11-18 19:02:34.917+0000 INFO  [neo4j/d68a0fec] [Source] Shutting down the Streams Source Module
2022-11-18 19:02:34.917+0000 INFO  [neo4j/d68a0fec] [Source] Initialising the Streams Source module
2022-11-18 19:02:34.934+0000 INFO  [neo4j/d68a0fec] Initialising Kafka Connector
2022-11-18 19:47:35.608+0000 INFO  [neo4j/d68a0fec] Kafka Connector started

the thing is, I know it should fail because I am giving the wrong port 🤔 but it doesn't fail.

Do you mind sharing a project with your setup and the expected operation to perform? that would be very helpful

[jameswylyreasint]

Yeah that's interesting.... Looking agian now it seems like our internal plugin was being re-included by the build process even though it'd deleted it.

My best guess at the moment is that something in there is causing these logs to be exposed but I still can't really tell where they're coming from.

I can't share our plugin so I'm trying to figure out which section of it is relevant for exposing these logs. I suspect it's something to do with Beeline which we use for cloud logging, which is taking these otherwise hidden logs and exposing them. AFAICT we don't have any other specialized logging config... maybe a log config in neo4j will allow these logs to be exposed again

[jameswylyreasint]

The warning message is unfortunate, but immaterial in the end. The real problem here is that the neo4j container cannot connect to the exposed kafka container even within the same docker network (for unknown reasons...)

I added some extra code to the test do use the admin kafka client to setup the topics, then try to use the neo4j streams plugin procedure to send a message to these created topics. Alongside the existing warnings (and errors about kafka timeouts) it produced a new error showing that the message could not be sent:

Trying to send a simple event with payload Sample message to kafka
Error while sending record to MyTopic1, because it doesn't exists

Example project link:
https://github.com/jameswylyreasint/neo4j-testcontainers-kafka-example

Updated to create topics via kafka admin client + use neo4j streams api via procedures to attempt to publish message to the created topic (confirmed via assertion, also confirmed via command line on running kafka container kafka-topics --list --bootstrap-server localhost:9092)

[eddumelendez]

Looks like we have been using the wrong port. If kafka:9092 is set to kafka.bootstrap.servers then it works as expected. assertThat(execResult.getStderr()).isEmpty(); should be commented.

[jameswylyreasint]

Yeah that is working now... I swear I had tried that initially but maybe there was something else wrong with the format at that point when I tested it.

Interestingly that is not present in any of the methods from testcontainers such as 'getExposedPorts', etc... Realistically this entire string should be returned from one of the methods in an obvious way.

Thanks