Is this intentional? ProducerFencedException doesn't inherit from RetriableException, so unless I'm missing something, this will still kill the thread. Same for below as well in StandbyTask.

Oh, boy. You're right... Really off my game today.

Why do we catch it at this level? Seems we should move it into ProcessorStateManger#flush() ?

Also note, that ProcessorStateManger#flush() only throws the first exception what might not be what we want to detect this case correctly.

Yeah, I wasn't sure... This way is a bit roundabout. On the other hand, right now, TaskMigratedException is only thrown by the tasks themselves or the threads that run them. It didn't seem within the responsibility of the ProcessorStateManager to reason about what specific exceptions imply about the state of the task.

Fair enough. It's not ideal that we get a ProcessorStateException and have to call getCause(), but well, it is what it is.

Reading the code here: we throw TaskMigratedException in both StreamTask as well as StreamThread. For the first case, the exception is thrown all the way to the caller of TaskManager and rethrown (hence in the level of StreamThread); the latter case is only a few exceptional situations where the tasks are closed due to underlying exceptions.

I feel throwing the exception in the level of StreamTask is fine, but we did too many capture / rethrow in different layers on top of it, which could be cleaned up in a separate tech-debt PR.

I very much agree. It's quite difficult to trace which exceptions would exist in different parts of the stack, and it would be nice to clean it up in the future.

@vvcephei while working on the cleanup PR I realized that when we call store.flush internally we should not blindly wrap all thrown exceptions from stores, but only those not StreamsException -- i.e. if the error is not from the store itself (like an IOException) but from the streams library, then just throw it out directly.

Will do this small refactoring in my PR.

The same argument as for GlobalStateStore should apply here -- StandbyTask don't write but only read, hence, no ProducerFencedException could ever happen here.

Oy. Good call. I've fixed it.

nit: e -> e.getCause()

Thanks! I didn't unwrap it in the log, since this preserves the full call stack (which helps track down why the exception happened)

I don't think we should log anything here -- we log and info level statement when we catch a TaskMigratedException and it's contains the full context as we pass in this and e -- otherwise, we would double log (also WARN does not seem to be appropriate anyway)

Roger that. I can remove it.

Why this change?

Sorry, I should have documented why I made another round of changes. When I tested the code prior to this change, we still had threads dying with a different exception...

[2019-11-27 11:28:59,965] ERROR [stream-soak-test-92665673-8057-4ee8-8619-bffd3670817d-StreamThread-2] stream-thread [stream-soak-test-92665673-8057-4ee8-8619-bffd3670817d-StreamThread-2] Encountered the following unexpected Kafka exception during processing, this usually indicate Streams internal errors: (org.apache.kafka.streams.processor.internals.StreamThread)
org.apache.kafka.streams.errors.StreamsException: stream-thread [stream-soak-test-92665673-8057-4ee8-8619-bffd3670817d-StreamThread-2] Failed to rebalance.
	at org.apache.kafka.streams.processor.internals.StreamThread.pollRequests(StreamThread.java:852)
	at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:739)
	at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:698)
	at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:671)
 Caused by: org.apache.kafka.streams.errors.StreamsException: stream-thread [stream-soak-test-92665673-8057-4ee8-8619-bffd3670817d-StreamThread-2] failed to suspend stream tasks
	at org.apache.kafka.streams.processor.internals.TaskManager.suspendActiveTasksAndState(TaskManager.java:253)
	at org.apache.kafka.streams.processor.internals.StreamsRebalanceListener.onPartitionsRevoked(StreamsRebalanceListener.java:116)
	at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.invokePartitionsRevoked(ConsumerCoordinator.java:291)
	at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onLeavePrepare(ConsumerCoordinator.java:707)
	at org.apache.kafka.clients.consumer.KafkaConsumer.unsubscribe(KafkaConsumer.java:1073)
	at org.apache.kafka.streams.processor.internals.StreamThread.enforceRebalance(StreamThread.java:716)
	at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:710)
	... 1 more
 Caused by: org.apache.kafka.streams.errors.TaskMigratedException: Client request for task 1_0 has been fenced due to a rebalance
	at org.apache.kafka.streams.processor.internals.AbstractTask.flushState(AbstractTask.java:182)
	at org.apache.kafka.streams.processor.internals.StreamTask.suspend(StreamTask.java:679)
	at org.apache.kafka.streams.processor.internals.StreamTask.close(StreamTask.java:787)
	at org.apache.kafka.streams.processor.internals.AssignedTasks.closeZombieTask(AssignedTasks.java:102)
	at org.apache.kafka.streams.processor.internals.AssignedStreamsTasks.suspendRunningTasks(AssignedStreamsTasks.java:151)
	at org.apache.kafka.streams.processor.internals.AssignedStreamsTasks.suspendOrCloseTasks(AssignedStreamsTasks.java:128)
	at org.apache.kafka.streams.processor.internals.TaskManager.suspendActiveTasksAndState(TaskManager.java:246)
	... 7 more
 Caused by: org.apache.kafka.streams.errors.ProcessorStateException: task [1_0] Failed to flush state store logData10MinuteFinalCount-store
	at org.apache.kafka.streams.processor.internals.ProcessorStateManager.flush(ProcessorStateManager.java:279)
	at org.apache.kafka.streams.processor.internals.AbstractTask.flushState(AbstractTask.java:178)
	... 13 more
 Caused by: org.apache.kafka.common.errors.ProducerFencedException: task [1_0] Abort sending since producer got fenced with a previous record (timestamp 1574853975939) to topic windowed-node-counts due to org.apache.kafka.common.errors.ProducerFencedException: Producer attempted an operation with an old epoch. Either there is a newer producer with the same transactionalId, or the producer's transaction has been expired by the broker.
 [2019-11-27 11:28:59,966] INFO [stream-soak-test-92665673-8057-4ee8-8619-bffd3670817d-StreamThread-2] stream-thread [stream-soak-test-92665673-8057-4ee8-8619-bffd3670817d-StreamThread-2] State transition from PARTITIONS_REVOKED to PENDING_SHUTDOWN (org.apache.kafka.streams.processor.internals.StreamThread)
 [2019-11-27 11:28:59,969] INFO [stream-soak-test-92665673-8057-4ee8-8619-bffd3670817d-StreamThread-2] stream-thread [stream-soak-test-92665673-8057-4ee8-8619-bffd3670817d-StreamThread-2] State transition from PENDING_SHUTDOWN to DEAD (org.apache.kafka.streams.processor.internals.StreamThread)

It turns out that firstException actually winds up being a fatal exception. The usages of this method express an intent to ignore any exceptions, so that's what this change does. I also made sure that any relevant state tracking still happens correctly. I.e., even if we can't cleanly close this zombie task, it's still removed from the "previous running" collection, etc.

I don't think that an UnknownProducerIdException implies that a task was migrated.

Agreed, but TaskMigratedException is the mechanism to initiate a rebalance, which is currently the only way that we can re-initialize the producers.

I think we can stand to re-evaluate how we're handling all these different producer and consumer exceptions (with and without eos). Right now, there are a lot of catch blocks for specific exceptions sprinkled throughout the codebase, and we universally either crash the thread or initiate a rebalance to attempt recovery. It would certainly benefit from developing a holistic approach, but right now, I'm just trying to get Streams threads to quit dying within a few hours of running in EOS mode, and I'm trying to restrain myself from broad refactoring so we can hope to merge this in for 2.4.0.

(About that last statement, I've been on holiday this week, so I haven't had time to document the comprehensive analysis in the ticket, but I do plan to make a case that it is a regression. When I ran 2.3 under similar conditions, we still had threads dying, but they were only dying from KIP-360-related-causes. In 2.4, Streams threads are dying from multiple new causes, related to the changes that we made in 2.4... Anyway, more details coming early next week, then we can discuss it)

Give the fix for the root cause of UnknownProducerIdException (via KIP-360), I am wondering if we should actually treat it as fatal? I also don't see why it would be regression if KS dies if this exception is thrown though? (I understand that we did introduce some regression bugs, but those seem unrelated to UnknownProducerIdException?)

To be fair, 2.4.0 does not contain a full implementation of KIP-360 and thus, this fix might be a workaround... But frankly, I am not very happy about it, even if I understand the desire to just stabilize it somehow without a refactoring that won't make it into 2.4.0 anyway.

Ok, then it seems worth hashing it out. Can you elaborate what is there to be unhappy about? I figured it would be reasonable to treat it as "recoverable by creating a new Producer" instead of "fatal and you should immediately terminate the thread".

What I was thinking was that is seems like the broker is telling us it doesn't know who we are (anymore). Because of the context, we know that it used to know who we were, so it must have forgotten somehow, and it doesn't really matter how because we can always shrug it off and re-create our producer to try again.

IIUC, KIP-360 would reduce the occurrence of the exception by keeping your producer id cached even after its ttl has expired, but it by no means guarantees that it'll remember you regardless of how long you're silent.

From that perspective it seems reasonable to catch this exception and conclude, "Oops, it looks like we were silent too long and our Producer has effectively expired. We should make a new one and try again." This isn't the same thing as getting fenced, but the resolution is the same (make a new Producer and try again).

Then again, this perspective may be based on a faulty understanding of the system. Is this approach unsafe, and we should actually terminate the application instead?

For this type or error, before KIP-360, the producer should self-recover and never throw an exception. (Cf "motivation" section of the KIP). If this error occurs, something really bad happened for the transaction and we are in an unknown state -- not sure if rebalancing and retrying is the right approach for this case \cc @guozhangwang @hachikuji @bob-barrett -- also, maybe it's just sufficient to close the task locally and recreate it -- I don't see a need to trigger a rebalance (in case it is ok to recorve from within KS).

This isn't the same thing as getting fenced, but the resolution is the same (make a new Producer and try again).

I disagree. If we got fenced, a new producer with the same transactional.id was created and we know that we don't own the task any longer. For a UnknownProducerIdException we still own the task.

Is this approach unsafe, and we should actually terminate the application instead?

I don't thinks it's unsafe, but it's not the "right" fix IMHO, and this bug is also not a regression.

Thanks for the clarification, @mjsax .

I agree that it would be better to just close the task and re-open it, but I don't feel comfortable introducing a whole new task lifecycle at this point for the 2.4.0 release. My rationale was that triggering a rebalance would put Kafka Streams through a well-known and well-tested code path that will result in the task getting closed and then opened again.

I do think it makes sense to go ahead and handle this case along with the two regressions. In my testing, this exception was just as fatal for Streams as the other two. Specifically, unless I include all three fixes, my soak test saw StreamThreads start dying within a few hours.

It seems like maybe a good approach right now would be to take the simple and sub-optimal path of just rebalancing when we encounter this exception for the 2.4.0 release, and file a Jira to optimize it in the way you suggest.

It would be good, by the way, to find out the answer to your question of whether it's safe to continue or not. Note that the current behavior, both in 2.4 and 2.3, without this patch is that a thread will get this exception and then shut itself down. This leads to a rebalance (since a member has left the group), after which the task is assigned to another thread, which continues processing it. Thus, we are already responding to this condition by rebalancing. It's just that we permanently lose a thread in the process. With this patch, we still rebalance, but we get to keep the thread running. If it's really unsafe to continue processing, though, we should stop the entire Streams application and force the operator to diagnose the problem with the brokers and start the app back up in a safe state.

I shared some thoughts on the comment above: before KIP-447, when we re-trigger the rebalance if the task is indeed migrated out the corresponding producer would be closed, otherwise that producer will be retained. But I think after KIP-447 we need to revisit this again.

At the moment I feel okay to treat UnknownProducerId equally as ProducerFenced.

Thanks. That will be a good opportunity to clean up what is honestly a pretty ham-handed approach here to gracefully deal with this condition without changing too much code.

Reading the code here: we throw TaskMigratedException in both StreamTask as well as StreamThread. For the first case, the exception is thrown all the way to the caller of TaskManager and rethrown (hence in the level of StreamThread); the latter case is only a few exceptional situations where the tasks are closed due to underlying exceptions.

I feel throwing the exception in the level of StreamTask is fine, but we did too many capture / rethrow in different layers on top of it, which could be cleaned up in a separate tech-debt PR.

For UnknownProducerIdException we actually can still reuse the client and just need to send the next produce request (unlike ProducerFencedException we cannot reuse the client and hence have to create a new client). However when we translate them into TaskMigrationException on higher level it would be treated equally as re-triggering a rebalance.

It's just that in current EOS implementation (before KIP-447) have one producer per task, hence when the task is closed the corresponding producer is closed and discarded as well. In KIP-447 however we may still need to treat them differently: for UnknownProducerIdException we can just log and proceed, for ProducerFencedException we need to close the producer and re-create a new one.

Thanks. One thing I'm not sure about, but felt at the time of putting this together was that, if we get an UnknownProducerIdException on one call, we'd probably just keep getting it on all subsequent calls. It seemed to me that the exception was originating from a condition where the broker "forgot" about our id, and if it has "forgotten" about us, I'm not sure why it would "remember" us later on.

If that's not true, then this approach certainly may result in more close+reopen operations than is necessary.

I shared some thoughts on the comment above: before KIP-447, when we re-trigger the rebalance if the task is indeed migrated out the corresponding producer would be closed, otherwise that producer will be retained. But I think after KIP-447 we need to revisit this again.

At the moment I feel okay to treat UnknownProducerId equally as ProducerFenced.

For here and producer.beginTransaction above: I think beginTxn / sendOffsetsToTxn / commitTxn would not throw UnknownProducerIdException but it does not harm to be more careful for 2.4 release -- so if you want to keep it as is I'm fine with it too.

Ok, I wasn't sure, so I just put it everywhere that we already check for ProducerFencedException, on the rationale that, if we can get fenced, then we must have a transactional id, and if we have an id, then it could be "unknown". Clearly, this is more intuitive than analytical, so I find your feedback plausible.

Minor for log4j succinctness: from looking at the production log files, I think keeping the whole stack trace of the original exception is not necessary since we know exactly it would be thrown from

	at org.apache.kafka.streams.processor.internals.RecordCollectorImpl$1.onCompletion(RecordCollectorImpl.java:202)
	at org.apache.kafka.clients.producer.KafkaProducer$InterceptorCallback.onCompletion(KafkaProducer.java:1318)
	at org.apache.kafka.clients.producer.internals.ProducerBatch.completeFutureAndFireCallbacks(ProducerBatch.java:230)
	at org.apache.kafka.clients.producer.internals.ProducerBatch.done(ProducerBatch.java:196)
	at org.apache.kafka.clients.producer.internals.Sender.failBatch(Sender.java:730)
	at org.apache.kafka.clients.producer.internals.Sender.failBatch(Sender.java:716)
	at org.apache.kafka.clients.producer.internals.Sender.completeBatch(Sender.java:674)
	at org.apache.kafka.clients.producer.internals.Sender.handleProduceResponse(Sender.java:596)
	at org.apache.kafka.clients.producer.internals.Sender.access$100(Sender.java:74)
	at org.apache.kafka.clients.producer.internals.Sender$1.onComplete(Sender.java:798)
	at org.apache.kafka.clients.ClientResponse.onComplete(ClientResponse.java:109)
	at org.apache.kafka.clients.NetworkClient.completeResponses(NetworkClient.java:561)
	at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:553)
	at org.apache.kafka.clients.producer.internals.Sender.runOnce(Sender.java:335)
	at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:244)

So may be we can just log the original exception name; similarly in recordSendError we can just keep the original exception's name but not the stack trace since we know exactly it is from the same Sender.failBatch trace.

In practice, the exceptions we're recording here are ApiExceptions, so they don't contain stack traces anyway.

I actually added the exception to the "cause" on purpose, because it was a pain trying to understand the stack traces we were previously logging while I was digging in to this issue. Even if the log messages are more verbose, they're easy to follow. I.e., it's a nice chain of "E1 caused by E2 caused by E3", instead of "E1-with-E3-in-the-message caused by E2".

I actually wanted to take it farther by removing the exception.toString from the message, but I didn't want to be too bold in this patch, and cause even more controversy.

SG!