Consumer doesn't shutdown when using Stage that terminates stream #166
Comments
|
thanks for reporting, @l15k4 |
|
Then there is a question in the air whether to explicitly stop & shutdown consumer stream when actor system terminates. My guess is that it shuts down implicitly on actor system termination but I've noticed of hanging kafka client-server connections if I didn't do this in my actors : override def postStop() = {
consumerControlOpt.foreach( control => control.stop.flatMap(_ => control.shutdown())(SameThreadExeC) )
}Wdyt? |
|
@l15k4 you can call shutdown directly. |
|
Hello, is there any update ? I am also facing this problem. |
|
@sasajib I am not sure what is expected behavior in this scenario. Current behavior is: we cancel The other one is to keep |
|
Last time I recommended to add a configurable timeout and then the consumer stops:
The timeout = 0 setting would reproduce the current behavior. |
|
@drewhk sounds like a plan |
|
Seems the issue is still there |
|
Confirmed that this issue exists with v0.14. In another class: Produces dead letter: |
|
Hello I suppose it is related to this issue - non terminating Consumer stream. How do we handle testing e.g. producer -> topic -> consumer |
|
Experiencing the dead letter problem too here. |
|
I am experiencing a related problem I believe: when a graph stage (such as mapAsync) inside a RunnableGraph fails with exception for which the decision should be Stop, it seems like the failure is not propagated somehow and consumer keeps running instead of shutting down, which keeps triggering. |
|
Experiencing the issue with
akka-stream-kafka 0.16 |
|
Im facing this problem too. Are there any workarounds? or Some sort of a dirty way of killing the consumer. I really need to kill it. |
|
@rabzu Would you have the chance to see if the problem is still there in 0.21? |
|
Still happens in 0.21: https://gitter.im/akka/reactive-kafka?at=5b1fb11f8864a936cc8167d2 |
|
Same issue here in 0.22 using |
|
This is a critical issue, we're not able to use the public class Main {
private static final String BOOTSTRAP_SERVERS = "kafka-server:9092";
private static final String IN_TOPIC = "in-topic";
private static final Logger LOGGER = LoggerFactory.getLogger(Main.class);
public static void main(String[] args) {
final ActorSystem system = ActorSystem.create("test");
final ActorMaterializer materializer = ActorMaterializer.create(
ActorMaterializerSettings.create(system).withSupervisionStrategy((Function<Throwable, Supervision.Directive>)(throwable -> {
LOGGER.error("Unexpected error in stream", throwable);
return Supervision.stop();
})),
system
);
final ConsumerSettings<byte[], byte[]> consumerSettings = ConsumerSettings
.create(consumerConfig, new ByteArrayDeserializer(), new ByteArrayDeserializer())
.withBootstrapServers(BOOTSTRAP_SERVERS)
.withGroupId("test-" + UUID.randomUUID())
.withClientId("test-" + UUID.randomUUID())
.withProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
.withProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
KillSwitch killSwitch = Consumer.committableSource(consumerSettings, Subscriptions.topics(IN_TOPIC))
.viaMat(KillSwitches.single(), Keep.right())
.mapAsync(30, msg -> {
// Simulate error
CompletableFuture<ConsumerMessage.CommittableMessage<byte[], byte[]>> future = new CompletableFuture<>();
future.completeExceptionally(new Exception("Error in mapAsync"));
return future;
})
.toMat(Sink.ignore(), Keep.left())
.run(materializer);
Runtime.getRuntime().addShutdownHook(new Thread(killSwitch::shutdown));
LOGGER.info("Await stream termination...");
try {
system.getWhenTerminated().toCompletableFuture().join();
// !! System never gets terminated
} catch (Exception ex) {
LOGGER.error("Unexpected error in stream", ex);
}
LOGGER.info("Shutting down stream...");
}
}The process never terminates, same issue if I call The only ugly workaround I've found to handle
private static final AtomicInteger EXIT_STATUS = new AtomicInteger(0);
...
final ActorMaterializer materializer = ActorMaterializer.create(
ActorMaterializerSettings.create(system).withSupervisionStrategy((Function<Throwable, Supervision.Directive>)(throwable -> {
LOGGER.error("Unexpected error in stream", throwable);
EXIT_STATUS.set(1);
System.exit(EXIT_STATUS.get());
return Supervision.stop();
})),
system
);
...
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
LOGGER.info("Shutting down stream...");
killSwitch.shutdown();
Runtime.getRuntime().halt(EXIT_STATUS.get());
})); |
|
The stream shuts down properly, including the |
|
Ok, I had misunderstood what was doing the KillSwitch: it ends the stream but not the system. Pair<UniqueKillSwitch, CompletionStage<Done>> pair = Consumer.committableSource(consumerSettings, Subscriptions.topics(IN_TOPIC))
.viaMat(KillSwitches.single(), Keep.right())
.mapAsync(30, msg -> {
// Simulate error
CompletableFuture<ConsumerMessage.CommittableMessage<byte[], byte[]>> future = new CompletableFuture<>();
future.completeExceptionally(new Exception("Error in mapAsync"));
return future;
})
.toMat(Sink.ignore(), Keep.both())
.run(materializer);
KillSwitch killSwitch = pair.first();
CompletionStage<Done> stage = pair.second();
Runtime.getRuntime().addShutdownHook(new Thread(killSwitch::shutdown));
LOGGER.info("Await stream termination...");
stage.whenComplete((done, throwable) -> {
if (throwable != null) {
LOGGER.error("Stream completed with an error", throwable);
}
LOGGER.info("Shutting down stream...");
system.terminate();
});Application now terminates smoothly as expected, thank you for the help. |
|
There hasn't been a documented case of the Actor not shutting down since v0.21. |
|
Hi, I think I am having the same issue with v1.0 I replicated a small piece of code that reproduces the issue: I am just sending 5 string messages to that Kafka topic, and when I stop the application, I get 2 different results:
I am assuming that the |
|
No, |
|
I'll close this now. The shutdown process has been reworked quite a bit since this was reported. Open a new issue if you find shutdown behaviour that is broken. |
Hey, I guess there is a bug in current Master branch related to stream termination. This stream :
yields
akka.kafka.KafkaConsumerActor$Internal$Stop$dead letter which imho leads to hangingKafkaConsumerActorwhich would be otherwise stopped ... As a result, if you want to compose streams like this :It gets stuck indefinitely in the second stream for some reason ...
The text was updated successfully, but these errors were encountered: