try to split context

Signed-off-by: tison <wander4096@gmail.com>
apache · Jul 17, 2023 · f8b7f55 · f8b7f55
1 parent 9ddf2fe
commit f8b7f55
Show file tree

Hide file tree

Showing 3 changed files with 58 additions and 12 deletions.
diff --git a/...-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSinkE2ECase.java b/...-pulsar-e2e-tests/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSinkE2ECase.java
@@ -51,7 +51,6 @@ public class PulsarSinkE2ECase extends SinkTestSuiteBase<String> {
     @TestExternalSystem
     PulsarContainerTestEnvironment pulsar = new PulsarContainerTestEnvironment(flink);
 
-    // Defines a set of external context Factories for different test cases.
     @TestContext
     PulsarTestContextFactory<String, SingleTopicProducingContext> sinkContext =
             new PulsarTestContextFactory<>(pulsar, SingleTopicProducingContext::new);

diff --git a/...ests/util/pulsar/PulsarSourceE2ECase.java → .../pulsar/PulsarSourceKeySharedE2ECase.java b/...ests/util/pulsar/PulsarSourceE2ECase.java → .../pulsar/PulsarSourceKeySharedE2ECase.java
@@ -18,9 +18,9 @@
 
 package org.apache.flink.tests.util.pulsar;
 
+import static org.apache.flink.streaming.api.CheckpointingMode.EXACTLY_ONCE;
 import org.apache.flink.connector.pulsar.testutils.PulsarTestContextFactory;
 import org.apache.flink.connector.pulsar.testutils.source.cases.MultipleTopicsConsumingContext;
-import org.apache.flink.connector.pulsar.testutils.source.cases.PartialKeysConsumingContext;
 import org.apache.flink.connector.testframe.junit.annotations.TestContext;
 import org.apache.flink.connector.testframe.junit.annotations.TestEnv;
 import org.apache.flink.connector.testframe.junit.annotations.TestExternalSystem;
@@ -29,15 +29,12 @@
 import org.apache.flink.streaming.api.CheckpointingMode;
 import org.apache.flink.tests.util.pulsar.common.FlinkContainerWithPulsarEnvironment;
 import org.apache.flink.tests.util.pulsar.common.PulsarContainerTestEnvironment;
-
 import org.junit.jupiter.api.Tag;
 
-import static org.apache.flink.streaming.api.CheckpointingMode.EXACTLY_ONCE;
-
 /** Pulsar source E2E test based on the connector testing framework. */
 @SuppressWarnings("unused")
 @Tag("org.apache.flink.testutils.junit.FailsOnJava11")
-public class PulsarSourceE2ECase extends SourceTestSuiteBase<String> {
+public class PulsarSourceKeySharedE2ECase extends SourceTestSuiteBase<String> {
 
     // Defines the Semantic.
     @TestSemantics CheckpointingMode[] semantics = new CheckpointingMode[] {EXACTLY_ONCE};
@@ -50,12 +47,7 @@ public class PulsarSourceE2ECase extends SourceTestSuiteBase<String> {
     @TestExternalSystem
     PulsarContainerTestEnvironment pulsar = new PulsarContainerTestEnvironment(flink);
 
-    // Defines a set of external context Factories for different test cases.
     @TestContext
-    PulsarTestContextFactory<String, MultipleTopicsConsumingContext> multipleTopic =
+    PulsarTestContextFactory<String, MultipleTopicsConsumingContext> context =
             new PulsarTestContextFactory<>(pulsar, MultipleTopicsConsumingContext::new);
-
-    @TestContext
-    PulsarTestContextFactory<String, PartialKeysConsumingContext> partialKeys =
-            new PulsarTestContextFactory<>(pulsar, PartialKeysConsumingContext::new);
 }
diff --git a/...s/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceMultipleTopicsE2ECase.java b/...s/src/test/java/org/apache/flink/tests/util/pulsar/PulsarSourceMultipleTopicsE2ECase.java
@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.tests.util.pulsar;
+
+import org.apache.flink.connector.pulsar.testutils.PulsarTestContextFactory;
+import org.apache.flink.connector.pulsar.testutils.source.cases.PartialKeysConsumingContext;
+import org.apache.flink.connector.testframe.junit.annotations.TestContext;
+import org.apache.flink.connector.testframe.junit.annotations.TestEnv;
+import org.apache.flink.connector.testframe.junit.annotations.TestExternalSystem;
+import org.apache.flink.connector.testframe.junit.annotations.TestSemantics;
+import org.apache.flink.connector.testframe.testsuites.SourceTestSuiteBase;
+import org.apache.flink.streaming.api.CheckpointingMode;
+import org.apache.flink.tests.util.pulsar.common.FlinkContainerWithPulsarEnvironment;
+import org.apache.flink.tests.util.pulsar.common.PulsarContainerTestEnvironment;
+
+import org.junit.jupiter.api.Tag;
+
+import static org.apache.flink.streaming.api.CheckpointingMode.EXACTLY_ONCE;
+
+/** Pulsar source E2E test based on the connector testing framework. */
+@SuppressWarnings("unused")
+@Tag("org.apache.flink.testutils.junit.FailsOnJava11")
+public class PulsarSourceMultipleTopicsE2ECase extends SourceTestSuiteBase<String> {
+
+    // Defines the Semantic.
+    @TestSemantics CheckpointingMode[] semantics = new CheckpointingMode[] {EXACTLY_ONCE};
+
+    // Defines TestEnvironment.
+    @TestEnv
+    FlinkContainerWithPulsarEnvironment flink = new FlinkContainerWithPulsarEnvironment(1, 6);
+
+    // Defines ConnectorExternalSystem.
+    @TestExternalSystem
+    PulsarContainerTestEnvironment pulsar = new PulsarContainerTestEnvironment(flink);
+
+    @TestContext
+    PulsarTestContextFactory<String, PartialKeysConsumingContext> context =
+            new PulsarTestContextFactory<>(pulsar, PartialKeysConsumingContext::new);
+}