Fix flakyness of testAutoSchemaFunction by creating the subscription before the function.
Otherwise if the function creates the subscription, there is no schema on the topic
and because of a bug a compatibility check fails on the new schema and the new consumer
can't be creatd
@@ -809,6 +809,11 @@ public abstract class PulsarFunctionsTest extends PulsarFunctionsTestBase {
String outputSchemaType,
SubscriptionInitialPosition subscriptionInitialPosition) throws Exception {
+ if (StringUtils.isNotEmpty(inputTopicName)) {
+ ensureSubscriptionCreated(
+ inputTopicName, String.format("public/default/%s", functionName), inputTopicSchema);
+ }
+
CommandGenerator generator;
log.info("------- INPUT TOPIC: '{}', customSchemaInputs: {}", inputTopicName, customSchemaInputs);
if (inputTopicName.endsWith(".*")) {
@@ -854,11 +859,6 @@ public abstract class PulsarFunctionsTest extends PulsarFunctionsTestBase {
ContainerExecResult result = pulsarCluster.getAnyWorker().execCmd(
commands);
assertTrue(result.getStdout().contains("Created successfully"));
-
- if (StringUtils.isNotEmpty(inputTopicName)) {
- ensureSubscriptionCreated(
- inputTopicName, String.format("public/default/%s", functionName), inputTopicSchema);
- }
}
private void updateFunctionParallelism(String functionName, int parallelism) throws Exception {
@@ -1536,6 +1536,8 @@ public abstract class PulsarFunctionsTest extends PulsarFunctionsTestBase {
String logTopicName,
String functionName,
Schema<?> schema) throws Exception {
+ ensureSubscriptionCreated(inputTopicName, String.format("public/default/%s", functionName), schema);
+
CommandGenerator generator;
log.info("------- INPUT TOPIC: '{}'", inputTopicName);
if (inputTopicName.endsWith(".*")) {
@@ -1556,8 +1558,6 @@ public abstract class PulsarFunctionsTest extends PulsarFunctionsTestBase {
ContainerExecResult result = pulsarCluster.getAnyWorker().execCmd(
commands);
assertTrue(result.getStdout().contains("Created successfully"));
-
- ensureSubscriptionCreated(inputTopicName, String.format("public/default/%s", functionName), schema);
}
private void publishAndConsumeMessages(String inputTopic,
