Skip to content

Fix compiler warning & thread was not terminated #137

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
Apr 24, 2018
Merged

Fix compiler warning & thread was not terminated #137

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
27dc5e3
Remove unused input parameter
jiminhsieh Apr 19, 2018
0a7702b
Fix thread was not terminated correctly.
jiminhsieh Apr 19, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions akka/src/test/scala/cakesolutions/kafka/akka/KafkaConsumerActorSpec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,7 @@ class KafkaConsumerActorSpec(system_ : ActorSystem) extends KafkaIntSpec(system_
""".stripMargin)
)

def configuredActor(topic: String): Config =
def configuredActor: Config =
ConfigFactory.parseString(
s"""
| bootstrap.servers = "localhost:$kafkaPort",
Expand Down Expand Up @@ -98,7 +98,7 @@ class KafkaConsumerActorSpec(system_ : ActorSystem) extends KafkaIntSpec(system_
producer.flush()

// Consumer and actor config in same config file
val consumer = system.actorOf(KafkaConsumerActor.props(configuredActor(topic), new StringDeserializer(), new StringDeserializer(), testActor))
val consumer = system.actorOf(KafkaConsumerActor.props(configuredActor, new StringDeserializer(), new StringDeserializer(), testActor))
consumer ! Subscribe.AutoPartition(List(topic))

val rs = expectMsgClass(30.seconds, classOf[ConsumerRecords[String, String]])
Expand All @@ -120,7 +120,7 @@ class KafkaConsumerActorSpec(system_ : ActorSystem) extends KafkaIntSpec(system_
val downstreamActor = TestProbe().ref

// Consumer and actor config in same config file
val consumer = system.actorOf(KafkaConsumerActor.props(configuredActor(topic), new StringDeserializer(), new StringDeserializer(), downstreamActor))
val consumer = system.actorOf(KafkaConsumerActor.props(configuredActor, new StringDeserializer(), new StringDeserializer(), downstreamActor))
consumer ! Subscribe.AutoPartition(List(topic))

// Initiate DeathWatch
Expand Down
6 changes: 4 additions & 2 deletions client/src/test/scala/cakesolutions/kafka/ConfigureSerializationSpec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,8 @@ class ConfigureSerializationSpec extends KafkaIntSpec{
""".stripMargin
), keySerializer, valueSerializer)

val _ = KafkaProducer(conf)
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Producer thread will stay in JVM even after finishing the test.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Here is the picture I recorded from VisualVM.
screen shot 2018-04-19 at 17 16 43

val producer = KafkaProducer(conf)
producer.close

keySerializer.configuration shouldEqual "mock_value"
keySerializer.isKeySerializer shouldEqual true
Expand All @@ -66,7 +67,8 @@ class ConfigureSerializationSpec extends KafkaIntSpec{
""".stripMargin
), keyDeserializer, valueDeserializer)

val _ = KafkaConsumer(conf)
val consumer = KafkaConsumer(conf)
consumer.close

keyDeserializer.configuration shouldEqual "mock_value"
keyDeserializer.isKeyDeserializer shouldEqual true
Expand Down