replication-offset-checkpoint.tmp (No such file or directory)

[MichalOrlowski]

We are using Embedded Kafka for Integration testing.
What we're getting is FileNotFoundException (sometimes) on Jenkins.
Question - is there a possibility to turn off ReplicaManager or so? Maybe there's another solution.

2016-10-13 07:53:35.161 ERROR 9469 --- [fka-scheduler-1] kafka.server.ReplicaManager              : [Replica Manager on Broker 0]: Error writing to highwatermark file: 

java.io.FileNotFoundException: /tmp/kafka-3266968904825284552/replication-offset-checkpoint.tmp (No such file or directory)
    at java.io.FileOutputStream.open0(Native Method)
    at java.io.FileOutputStream.open(FileOutputStream.java:270)
    at java.io.FileOutputStream.<init>(FileOutputStream.java:213)
    at java.io.FileOutputStream.<init>(FileOutputStream.java:162)
    at kafka.server.OffsetCheckpoint.write(OffsetCheckpoint.scala:37)
    at kafka.server.ReplicaManager$$anonfun$checkpointHighWatermarks$2.apply(ReplicaManager.scala:874)
    at kafka.server.ReplicaManager$$anonfun$checkpointHighWatermarks$2.apply(ReplicaManager.scala:871)
    at scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:778)
    at scala.collection.immutable.Map$Map1.foreach(Map.scala:116)
    at scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:777)
    at kafka.server.ReplicaManager.checkpointHighWatermarks(ReplicaManager.scala:871)
    at kafka.server.ReplicaManager$$anonfun$1.apply$mcV$sp(ReplicaManager.scala:153)
    at kafka.utils.KafkaScheduler$$anonfun$1.apply$mcV$sp(KafkaScheduler.scala:110)
    at kafka.utils.CoreUtils$$anon$1.run(CoreUtils.scala:60)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)

[artembilan]

Doesn't that mean that you have a low ulimit on that server?
https://easyengine.io/tutorials/linux/increase-open-files-limit/

[MichalOrlowski]

No, that's not it. I tried to increase it, but still got same problem.

[artembilan]

Well, share, please, as much info then as possible: test-case to play from our side, how much concurrent tests you have, aren't there some process on that server which clean the /tmp dir, which Kafka version do you use, etc. etc..

Looks like it is more Apache Kafka issue than Spring Kafka...
I'm afraid that we can't switch off ReplicaManager...

[MichalOrlowski]

I've noticed that this issue occures when Jenkins builds two microservices simultaneously (and both are using Kafka Test). I think first build cleans up and second one cannot access it.
/tmp file is Jenkins global directory (not per build).

Question - is there a way to customize /tmp dir name? We need to have possibility building services simultaneously.

[artembilan]

As I said before: it is the question directly to Apache Kafka.

public EmbeddedZookeeper() {
        this.snapshotDir = kafka.utils.TestUtils..MODULE$.tempDir();
        this.logDir = kafka.utils.TestUtils..MODULE$.tempDir();

I'm not familiar with Scala, so can't debug and interpret what they do there...

[garyrussell]

It seems unlikely to be related to concurrent builds; in my experience, the /kafka-n/ directory gets a different n on each run.

Yeah - it uses Files.createTempDirectory() which uses random.nextLong() where random is a new SecureRandom so it's pretty unlikely that there would be a collision.

[MichalOrlowski]

Isn't ReplicaManager.checkpointHighWatermarks() method clearing all available replicas?

I think there's possibility to remove /tmp/kafka-n by other thread.

@see core/src/main/scala/kafka/server/ReplicaManager.scala

[artembilan]

I still don't understand why do you ask that question to us, Spring community when you see the problem in the Apache Kafka directly?
Let's move this discussion to StackOverflow: there is nothing from the Spring perspective to fix or improve.

If you insist it is ReplicaManager, so please go to the Apache Kafka community and ask there.
That is so low level of the integration that we are just not aware of.

Sorry, but it looks like we (at least me) are useless for your on the topic and I don't understand why do you spend time with us not Apache Kafka community?

Would be glad to see some cross-link from there to widen knowledge in this Kafka topic.

Thanks

[MichalOrlowski]

Right now I just mocked ReplicaManager.shutdown() method to return false.

Thanks for you comments.

[hoaz]

I have found this thread via google looking for solution. Did not find it here, but this is the only place which refers to problem I encountered.
I did analysis and want to share it with others. Here is what happens:

1. kafka TestUtils.tempDirectory method is used to create temporary directory for embedded kafka broker. It also registers shutdown hook which deletes this directory when JVM exits.
2. when unit test finishes execution it calls System.exit, which in turn executes all registered shutdown hooks

If kafka broker runs at the end of unit test it will attempt to write/read data in a dir which is deleted and produces different FileNotFound exceptions.

Solution: shutdown embedded kafka broker at the end of the test before System.exit is called.

If KafkaEmbedded rule is used properly it will call KafkaEmbedded#after method, which destroys broker before System#exit is called.

I use KafkaEmbedded class in Spring integration test and create it as bean. Unfortunately spring context is destroyed in shutdown hook as well and it happens concurrently with other shutdown hooks, so kafka log directory is destroyed before embedded kafka broker is down. I did not find proper solution yet for this usage scenario.

[artembilan]

@hoaz ,

thank you for your report.
The question is: why do you use KafkaEmbedded as a bean?
Why @ClassRule is not sufficient for you?

@ClassRule
public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, INT_KEY_TOPIC, STRING_KEY_TOPIC);

If kafka broker runs at the end of unit test it will attempt to write/read data in a dir which is deleted and produces different FileNotFound exceptions.

Well, it would be better to stop writing in the end of test, e.g. call stop() of the endpoints or verify that sent message has been reached broker before existing from the test. Right?

[hoaz]

Because I have multiple integration tests which share the same Spring test context. I do not want to bring up and shut down kafka broker in each test but rather delegate it to Spring, which caches spring context between tests.
This approach speeds up integration tests with common test Spring context significantly.

[artembilan]

Would you mind to share the code how do you do that?
According @ClassRule JavaDocs we have:

* For example, here is a test suite that connects to a server once before
 * all the test classes run, and disconnects after they are finished:
 * <pre>
 * &#064;RunWith(Suite.class)
 * &#064;SuiteClasses({A.class, B.class, C.class})
 * public class UsesExternalResource {
 *     public static Server myServer= new Server();
 *
 *     &#064;ClassRule
 *     public static ExternalResource resource= new ExternalResource() {

So, placing the KafkaEmbedded rule there on a suite level, should make us happy. Isn't it?