[KAFKA-15690] EosIntegrationTest is flaky.

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Component/s: streams, unit tests
• Labels: flaky-test
• Assignee: Lucas Brutschy
• Reporter: Calvin Liu
• Created: 26/Oct/23 15:52
• Updated: 19/Feb/24 09:19
• Resolved: 04/Dec/23 08:50

Description

EosIntegrationTest

shouldNotViolateEosIfOneTaskGetsFencedUsingIsolatedAppInstances[exactly_once_v2, processing threads = false]

org.junit.runners.model.TestTimedOutException: test timed out after 600 seconds	at org.apache.kafka.clients.NetworkClient$DefaultMetadataUpdater.handleServerDisconnect(NetworkClient.java:1111)	at org.apache.kafka.clients.NetworkClient.processDisconnection(NetworkClient.java:821)	at org.apache.kafka.clients.NetworkClient.processTimeoutDisconnection(NetworkClient.java:779)	at org.apache.kafka.clients.NetworkClient.handleTimedOutRequests(NetworkClient.java:837)

org.apache.kafka.streams.errors.StreamsException: Exception caught in process. taskId=0_1, processor=KSTREAM-SOURCE-0000000000, topic=multiPartitionInputTopic, partition=1, offset=15, stacktrace=java.lang.RuntimeException: Detected we've been interrupted.	at org.apache.kafka.streams.integration.EosIntegrationTest$1$1.transform(EosIntegrationTest.java:892)	at org.apache.kafka.streams.integration.EosIntegrationTest$1$1.transform(EosIntegrationTest.java:867)	at org.apache.kafka.streams.kstream.internals.TransformerSupplierAdapter$1.transform(TransformerSupplierAdapter.java:49)	at org.apache.kafka.streams.kstream.internals.TransformerSupplierAdapter$1.transform(TransformerSupplierAdapter.java:38)	at org.apache.kafka.streams.kstream.internals.KStreamFlatTransform$KStreamFlatTransformProcessor.process(KStreamFlatTransform.java:66) 

  shouldWriteLatestOffsetsToCheckpointOnShutdown[exactly_once_v2, processing threads = false] 

java.lang.RuntimeException: java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.UnknownServerException: The server experienced an unexpected error when processing the request.	at org.apache.kafka.streams.integration.utils.KafkaEmbedded.deleteTopic(KafkaEmbedded.java:204)	at org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster.deleteTopicsAndWait(EmbeddedKafkaCluster.java:286)	at org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster.deleteTopicsAndWait(EmbeddedKafkaCluster.java:274)	at org.apache.kafka.streams.integration.EosIntegrationTest.createTopics(EosIntegrationTest.java:174)	at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:103)

org.apache.kafka.streams.errors.StreamsException: Exception caught in process. taskId=0_1, processor=KSTREAM-SOURCE-0000000000, topic=multiPartitionInputTopic, partition=1, offset=15, stacktrace=java.lang.RuntimeException: Detected we've been interrupted.	at org.apache.kafka.streams.integration.EosIntegrationTest$1$1.transform(EosIntegrationTest.java:892)	at org.apache.kafka.streams.integration.EosIntegrationTest$1$1.transform(EosIntegrationTest.java:867)	at org.apache.kafka.streams.kstream.internals.TransformerSupplierAdapter$1.transform(TransformerSupplierAdapter.java:49)	at org.apache.kafka.streams.kstream.internals.TransformerSupplierAdapter$1.transform(TransformerSupplierAdapter.java:38)	at org.apache.kafka.streams.kstream.internals.KStreamFlatTransform$KStreamFlatTransformProcessor.process(KStreamFlatTransform.java:66) 

shouldWriteLatestOffsetsToCheckpointOnShutdown[exactly_once, processing threads = false] 

org.opentest4j.AssertionFailedError: Condition not met within timeout 60000. StreamsTasks did not request commit. ==> expected: <true> but was: <false>	at app//org.junit.jupiter.api.AssertionFailureBuilder.build(AssertionFailureBuilder.java:151)	at app//org.junit.jupiter.api.AssertionFailureBuilder.buildAndThrow(AssertionFailureBuilder.java:132)	at app//org.junit.jupiter.api.AssertTrue.failNotTrue(AssertTrue.java:63)	at app//org.junit.jupiter.api.AssertTrue.assertTrue(AssertTrue.java:36)	at app//org.junit.jupiter.api.Assertions.assertTrue(Assertions.java:210)

java.lang.IllegalStateException: Replica [Topic=__transaction_state,Partition=2,Replica=1] should be in the OfflineReplica,ReplicaDeletionStarted states before moving to ReplicaDeletionIneligible state. Instead it is in OnlineReplica state	at kafka.controller.ZkReplicaStateMachine.logInvalidTransition(ReplicaStateMachine.scala:442)	at kafka.controller.ZkReplicaStateMachine.$anonfun$doHandleStateChanges$2(ReplicaStateMachine.scala:164)	at kafka.controller.ZkReplicaStateMachine.$anonfun$doHandleStateChanges$2$adapted(ReplicaStateMachine.scala:164)	at scala.collection.immutable.List.foreach(List.scala:333) 

They are running long and may relate to timeout.

Issue Links

• links to: GitHub Pull Request #14869