[FLINK-25522] KafkaShuffleExactlyOnceITCase.testAssignedToPartitionFailureRecoveryProcessingTime

• Type: Sub-task
• Status: Closed
• Priority: Critical
• Resolution: Duplicate
• Affects Version/s: 1.15.0
• Fix Version/s: 1.15.0
• Component/s: Connectors / Kafka
• Labels: test-stability
• Assignee: Unassigned
• Reporter: Till Rohrmann
• Created: 05/Jan/22 08:31
• Updated: 21/Feb/22 17:05
• Resolved: 21/Feb/22 17:05

Description

The test KafkaShuffleExactlyOnceITCase.testAssignedToPartitionFailureRecoveryProcessingTime failed on AZP with:

2022-01-05T04:31:25.7208273Z java.util.concurrent.TimeoutException: The topic metadata failed to propagate to Kafka broker.
2022-01-05T04:31:25.7210543Z 	at org.apache.flink.core.testutils.CommonTestUtils.waitUtil(CommonTestUtils.java:214)
2022-01-05T04:31:25.7211289Z 	at org.apache.flink.core.testutils.CommonTestUtils.waitUtil(CommonTestUtils.java:230)
2022-01-05T04:31:25.7212025Z 	at org.apache.flink.streaming.connectors.kafka.KafkaTestEnvironmentImpl.createTestTopic(KafkaTestEnvironmentImpl.java:216)
2022-01-05T04:31:25.7212944Z 	at org.apache.flink.streaming.connectors.kafka.KafkaTestEnvironment.createTestTopic(KafkaTestEnvironment.java:98)
2022-01-05T04:31:25.7213794Z 	at org.apache.flink.streaming.connectors.kafka.KafkaTestBase.createTestTopic(KafkaTestBase.java:216)
2022-01-05T04:31:25.7214854Z 	at org.apache.flink.streaming.connectors.kafka.shuffle.KafkaShuffleExactlyOnceITCase.testAssignedToPartitionFailureRecovery(KafkaShuffleExactlyOnceITCase.java:158)
2022-01-05T04:31:25.7215823Z 	at org.apache.flink.streaming.connectors.kafka.shuffle.KafkaShuffleExactlyOnceITCase.testAssignedToPartitionFailureRecoveryProcessingTime(KafkaShuffleExactlyOnceITCase.java:81)
2022-01-05T04:31:25.7216532Z 	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
2022-01-05T04:31:25.7217307Z 	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
2022-01-05T04:31:25.7217917Z 	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
2022-01-05T04:31:25.7218437Z 	at java.lang.reflect.Method.invoke(Method.java:498)
2022-01-05T04:31:25.7218969Z 	at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:59)
2022-01-05T04:31:25.7219572Z 	at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
2022-01-05T04:31:25.7220183Z 	at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:56)
2022-01-05T04:31:25.7220770Z 	at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
2022-01-05T04:31:25.7221346Z 	at org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
2022-01-05T04:31:25.7221959Z 	at org.junit.internal.runners.statements.FailOnTimeout$CallableStatement.call(FailOnTimeout.java:299)
2022-01-05T04:31:25.7222603Z 	at org.junit.internal.runners.statements.FailOnTimeout$CallableStatement.call(FailOnTimeout.java:293)
2022-01-05T04:31:25.7223413Z 	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
2022-01-05T04:31:25.7223871Z 	at java.lang.Thread.run(Thread.java:748)
2022-01-05T04:31:25.7321823Z java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TopicExistsException: Topic 'partition_failure_recovery_ProcessingTime' already exists.
2022-01-05T04:31:25.7323411Z 	at org.apache.kafka.common.internals.KafkaFutureImpl.wrapAndThrow(KafkaFutureImpl.java:45)
2022-01-05T04:31:25.7324069Z 	at org.apache.kafka.common.internals.KafkaFutureImpl.access$000(KafkaFutureImpl.java:32)
2022-01-05T04:31:25.7324696Z 	at org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:89)
2022-01-05T04:31:25.7325309Z 	at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:260)
2022-01-05T04:31:25.7326077Z 	at org.apache.flink.streaming.connectors.kafka.KafkaTestEnvironmentImpl.createTestTopic(KafkaTestEnvironmentImpl.java:214)
2022-01-05T04:31:25.7326999Z 	at org.apache.flink.streaming.connectors.kafka.KafkaTestEnvironment.createTestTopic(KafkaTestEnvironment.java:98)
2022-01-05T04:31:25.7327659Z 	at org.apache.flink.streaming.connectors.kafka.KafkaTestBase.createTestTopic(KafkaTestBase.java:216)
2022-01-05T04:31:25.7328418Z 	at org.apache.flink.streaming.connectors.kafka.shuffle.KafkaShuffleExactlyOnceITCase.testAssignedToPartitionFailureRecovery(KafkaShuffleExactlyOnceITCase.java:158)
2022-01-05T04:31:25.7329328Z 	at org.apache.flink.streaming.connectors.kafka.shuffle.KafkaShuffleExactlyOnceITCase.testAssignedToPartitionFailureRecoveryProcessingTime(KafkaShuffleExactlyOnceITCase.java:81)
2022-01-05T04:31:25.7330013Z 	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
2022-01-05T04:31:25.7330507Z 	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
2022-01-05T04:31:25.7331082Z 	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
2022-01-05T04:31:25.7331604Z 	at java.lang.reflect.Method.invoke(Method.java:498)
2022-01-05T04:31:25.7332098Z 	at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:59)
2022-01-05T04:31:25.7332672Z 	at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
2022-01-05T04:31:25.7333491Z 	at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:56)
2022-01-05T04:31:25.7334065Z 	at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
2022-01-05T04:31:25.7334623Z 	at org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
2022-01-05T04:31:25.7335205Z 	at org.junit.internal.runners.statements.FailOnTimeout$CallableStatement.call(FailOnTimeout.java:299)
2022-01-05T04:31:25.7335855Z 	at org.junit.internal.runners.statements.FailOnTimeout$CallableStatement.call(FailOnTimeout.java:293)
2022-01-05T04:31:25.7336400Z 	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
2022-01-05T04:31:25.7336935Z 	at java.lang.Thread.run(Thread.java:748)
2022-01-05T04:31:25.7337809Z Caused by: org.apache.kafka.common.errors.TopicExistsException: Topic 'partition_failure_recovery_ProcessingTime' already exists.
2022-01-05T04:31:25.7406273Z java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TopicExistsException: Topic 'partition_failure_recovery_ProcessingTime' already exists.
2022-01-05T04:31:25.7407400Z 	at org.apache.kafka.common.internals.KafkaFutureImpl.wrapAndThrow(KafkaFutureImpl.java:45)
2022-01-05T04:31:25.7408341Z 	at org.apache.kafka.common.internals.KafkaFutureImpl.access$000(KafkaFutureImpl.java:32)
2022-01-05T04:31:25.7408983Z 	at org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:89)
2022-01-05T04:31:25.7409576Z 	at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:260)
2022-01-05T04:31:25.7410212Z 	at org.apache.flink.streaming.connectors.kafka.KafkaTestEnvironmentImpl.createTestTopic(KafkaTestEnvironmentImpl.java:214)
2022-01-05T04:31:25.7410924Z 	at org.apache.flink.streaming.connectors.kafka.KafkaTestEnvironment.createTestTopic(KafkaTestEnvironment.java:98)
2022-01-05T04:31:25.7411572Z 	at org.apache.flink.streaming.connectors.kafka.KafkaTestBase.createTestTopic(KafkaTestBase.java:216)
2022-01-05T04:31:25.7412324Z 	at org.apache.flink.streaming.connectors.kafka.shuffle.KafkaShuffleExactlyOnceITCase.testAssignedToPartitionFailureRecovery(KafkaShuffleExactlyOnceITCase.java:158)
2022-01-05T04:31:25.7413408Z 	at org.apache.flink.streaming.connectors.kafka.shuffle.KafkaShuffleExactlyOnceITCase.testAssignedToPartitionFailureRecoveryProcessingTime(KafkaShuffleExactlyOnceITCase.java:81)
2022-01-05T04:31:25.7414097Z 	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
2022-01-05T04:31:25.7414590Z 	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
2022-01-05T04:31:25.7415166Z 	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
2022-01-05T04:31:25.7415689Z 	at java.lang.reflect.Method.invoke(Method.java:498)
2022-01-05T04:31:25.7416349Z 	at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:59)
2022-01-05T04:31:25.7416910Z 	at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
2022-01-05T04:31:25.7417480Z 	at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:56)
2022-01-05T04:31:25.7418037Z 	at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
2022-01-05T04:31:25.7418617Z 	at org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
2022-01-05T04:31:25.7419207Z 	at org.junit.internal.runners.statements.FailOnTimeout$CallableStatement.call(FailOnTimeout.java:299)
2022-01-05T04:31:25.7419840Z 	at org.junit.internal.runners.statements.FailOnTimeout$CallableStatement.call(FailOnTimeout.java:293)
2022-01-05T04:31:25.7420387Z 	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
2022-01-05T04:31:25.7420821Z 	at java.lang.Thread.run(Thread.java:748)
2022-01-05T04:31:25.7421652Z Caused by: org.apache.kafka.common.errors.TopicExistsException: Topic 'partition_failure_recovery_ProcessingTime' already exists.
2022-01-05T04:31:25.7481712Z java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TopicExistsException: Topic 'partition_failure_recovery_ProcessingTime' already exists.
2022-01-05T04:31:25.7482663Z 	at org.apache.kafka.common.internals.KafkaFutureImpl.wrapAndThrow(KafkaFutureImpl.java:45)
2022-01-05T04:31:25.7483460Z 	at org.apache.kafka.common.internals.KafkaFutureImpl.access$000(KafkaFutureImpl.java:32)
2022-01-05T04:31:25.7484070Z 	at org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:89)
2022-01-05T04:31:25.7484661Z 	at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:260)
2022-01-05T04:31:25.7485296Z 	at org.apache.flink.streaming.connectors.kafka.KafkaTestEnvironmentImpl.createTestTopic(KafkaTestEnvironmentImpl.java:214)
2022-01-05T04:31:25.7486170Z 	at org.apache.flink.streaming.connectors.kafka.KafkaTestEnvironment.createTestTopic(KafkaTestEnvironment.java:98)
2022-01-05T04:31:25.7486831Z 	at org.apache.flink.streaming.connectors.kafka.KafkaTestBase.createTestTopic(KafkaTestBase.java:216)
2022-01-05T04:31:25.7487572Z 	at org.apache.flink.streaming.connectors.kafka.shuffle.KafkaShuffleExactlyOnceITCase.testAssignedToPartitionFailureRecovery(KafkaShuffleExactlyOnceITCase.java:158)
2022-01-05T04:31:25.7488482Z 	at org.apache.flink.streaming.connectors.kafka.shuffle.KafkaShuffleExactlyOnceITCase.testAssignedToPartitionFailureRecoveryProcessingTime(KafkaShuffleExactlyOnceITCase.java:81)
2022-01-05T04:31:25.7489241Z 	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
2022-01-05T04:31:25.7489717Z 	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
2022-01-05T04:31:25.7490270Z 	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
2022-01-05T04:31:25.7490770Z 	at java.lang.reflect.Method.invoke(Method.java:498)
2022-01-05T04:31:25.7491261Z 	at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:59)
2022-01-05T04:31:25.7491819Z 	at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
2022-01-05T04:31:25.7492350Z 	at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:56)
2022-01-05T04:31:25.7492972Z 	at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
2022-01-05T04:31:25.7493560Z 	at org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
2022-01-05T04:31:25.7494119Z 	at org.junit.internal.runners.statements.FailOnTimeout$CallableStatement.call(FailOnTimeout.java:299)
2022-01-05T04:31:25.7494722Z 	at org.junit.internal.runners.statements.FailOnTimeout$CallableStatement.call(FailOnTimeout.java:293)
2022-01-05T04:31:25.7495244Z 	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
2022-01-05T04:31:25.7495667Z 	at java.lang.Thread.run(Thread.java:748)
2022-01-05T04:31:25.7496543Z Caused by: org.apache.kafka.common.errors.TopicExistsException: Topic 'partition_failure_recovery_ProcessingTime' already exists.
2022-01-05T04:32:29.6600468Z Jan 05 04:32:29 [ERROR] Tests run: 6, Failures: 1, Errors: 0, Skipped: 0, Time elapsed: 119.592 s <<< FAILURE! - in org.apache.flink.streaming.connectors.kafka.shuffle.KafkaShuffleExactlyOnceITCase
2022-01-05T04:32:29.6602386Z Jan 05 04:32:29 [ERROR] testAssignedToPartitionFailureRecoveryProcessingTime  Time elapsed: 21.118 s  <<< FAILURE!
2022-01-05T04:32:29.6604407Z Jan 05 04:32:29 java.lang.AssertionError: Create test topic : partition_failure_recovery_ProcessingTime failed, org.apache.kafka.common.errors.TopicExistsException: Topic 'partition_failure_recovery_ProcessingTime' already exists.
2022-01-05T04:32:29.6605353Z Jan 05 04:32:29 	at org.junit.Assert.fail(Assert.java:89)
2022-01-05T04:32:29.6605990Z Jan 05 04:32:29 	at org.apache.flink.streaming.connectors.kafka.KafkaTestEnvironmentImpl.createTestTopic(KafkaTestEnvironmentImpl.java:223)
2022-01-05T04:32:29.6606821Z Jan 05 04:32:29 	at org.apache.flink.streaming.connectors.kafka.KafkaTestEnvironment.createTestTopic(KafkaTestEnvironment.java:98)
2022-01-05T04:32:29.6607531Z Jan 05 04:32:29 	at org.apache.flink.streaming.connectors.kafka.KafkaTestBase.createTestTopic(KafkaTestBase.java:216)
2022-01-05T04:32:29.6608326Z Jan 05 04:32:29 	at org.apache.flink.streaming.connectors.kafka.shuffle.KafkaShuffleExactlyOnceITCase.testAssignedToPartitionFailureRecovery(KafkaShuffleExactlyOnceITCase.java:158)
2022-01-05T04:32:29.6609275Z Jan 05 04:32:29 	at org.apache.flink.streaming.connectors.kafka.shuffle.KafkaShuffleExactlyOnceITCase.testAssignedToPartitionFailureRecoveryProcessingTime(KafkaShuffleExactlyOnceITCase.java:81)
2022-01-05T04:32:29.6610010Z Jan 05 04:32:29 	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
2022-01-05T04:32:29.6610904Z Jan 05 04:32:29 	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
2022-01-05T04:32:29.6611777Z Jan 05 04:32:29 	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
2022-01-05T04:32:29.6612352Z Jan 05 04:32:29 	at java.lang.reflect.Method.invoke(Method.java:498)
2022-01-05T04:32:29.6619881Z Jan 05 04:32:29 	at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:59)
2022-01-05T04:32:29.6620644Z Jan 05 04:32:29 	at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
2022-01-05T04:32:29.6621272Z Jan 05 04:32:29 	at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:56)
2022-01-05T04:32:29.6621886Z Jan 05 04:32:29 	at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
2022-01-05T04:32:29.6622486Z Jan 05 04:32:29 	at org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
2022-01-05T04:32:29.6623365Z Jan 05 04:32:29 	at org.junit.internal.runners.statements.FailOnTimeout$CallableStatement.call(FailOnTimeout.java:299)
2022-01-05T04:32:29.6624203Z Jan 05 04:32:29 	at org.junit.internal.runners.statements.FailOnTimeout$CallableStatement.call(FailOnTimeout.java:293)
2022-01-05T04:32:29.6624983Z Jan 05 04:32:29 	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
2022-01-05T04:32:29.6625612Z Jan 05 04:32:29 	at java.lang.Thread.run(Thread.java:748)
2022-01-05T04:32:29.6626174Z Jan 05 04:32:29 

https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=28946&view=logs&j=c5f0071e-1851-543e-9a45-9ac140befc32&t=15a22db7-8faa-5b34-3920-d33c9f0ca23c&l=7501

Issue Links

• duplicates: FLINK-26115 Multiple Kafka connector tests failed due to The topic metadata failed to propagate to Kafka broker