Concurrent kafka instances #105
Merged
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
added 3 commits
February 18, 2018 20:06
…in case class instances. The start methods, which used to return Unit, now return references to the base type: EmbeddedServer. These references can be used to stop the specific servers using a new overload stop function.
|
Thanks @tjheslin1 - the approach looks good to me. Let me know when you've finished and I'll carry on with the review. |
… kafka functionality after first has shutdown
tjheslin1
changed the title
|
Ready to be reviewed. |
|
All looks good - could you please address the 2 minor things that are still pending? |
tjheslin1
force-pushed
the
concurrent-kafka
branch
from
0aeeee4 to
1685403
Compare
|
Ready now @manub. @Shashidesai will be raising another issue. |
|
Thanks. Will merge now and release in the next days. |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
I'm raising this PR now in the hopes that you can take a look at my approach. If you have any major concerns it would be best to find out sooner!
Closes #101
TODO:
runStreamsimplicit conversionextension method to go fromSeq[EmbeddedServer]=>Seq[EmbeddedZ]orSeq[EmbeddedK], replacing the boiler placeisInstanceOf+asInstanceOfEmbeddedServer.scalaProblem
Running tests in parallel which use
EmbeddedKafkacan run into an issue with the log files asEmbeddedKafka#stopclears all temporary log directories, which another instance of kafka may still be relying on in another test.To see this error I changed
logback.xmlline 1 to:log4j.rootLogger=error,stdout:FATAL [Replica Manager on Broker 0]: Error writing to highwatermark file...java.io.FileNotFoundException...kafka-logsXXXXXXXXX.tmp/replication-offset-checkpoint.tmp (No such file or directory)Testing
I would like to improve these as there is quite a lot of repetition
Solution
I have created a model of the running servers, both Kafka and Zookeeper. Starting one of these appends an
EmbeddedServerto managed list ofservers.The following functions have had their signatures modified. Given that they all used to return
Unit, this results in a non-breaking change to users. On upgrading the returned objects will be ignored and the existing functionality will be preserved.EmbeddedKafka#start now returns an
EmbeddedServer.EmbeddedKafka#startZooKeeper now returns an
EmbeddedZ.EmbeddedKafka#startKafka now returns an
EmbeddedK.I have added an overload to EmbeddedKafka#stop which will stop the specific server and remove it from the list of managed servers.