[FLINK-25220][test] Split architecture-tests and build ArchUnit exten…

…sion

- create submodule flink-architecture-tests-base, flink-architecture-tests-production, flink-architecture-tests-test
- build ArchUnit extension and move the common part to the flink-architecture-tests-base submodule
- refactoring and move ArchUnit tests and rules to the flink-architecture-tests-production submodule

flink-architecture-tests/flink-architecture-tests-base/pom.xml

@@ -0,0 +1,46 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+
+	<modelVersion>4.0.0</modelVersion>
+
+	<parent>
+		<groupId>org.apache.flink</groupId>
+		<artifactId>flink-architecture-tests</artifactId>
+		<version>1.15-SNAPSHOT</version>
+		<relativePath>../pom.xml</relativePath>
+	</parent>
+
+	<artifactId>flink-architecture-tests-base</artifactId>
+	<name>Flink : Architecture Tests : Base</name>
+
+	<packaging>jar</packaging>
+
+	<dependencies>
+		<dependency>
+			<groupId>com.tngtech.archunit</groupId>
+			<artifactId>archunit</artifactId>
+			<scope>compile</scope>
+		</dependency>
+	</dependencies>
+
+</project>

flink-architecture-tests/src/test/java/org/apache/flink/architecture/ArchitectureTest.java → flink-architecture-tests/flink-architecture-tests-base/src/main/java/org/apache/flink/architecture/common/ImportOptions.java

@@ -16,36 +16,15 @@
  * limitations under the License.
  */
 
-package org.apache.flink.architecture;
-
-import org.apache.flink.architecture.common.GivenJavaClasses;
-import org.apache.flink.architecture.common.SourcePredicates;
-import org.apache.flink.architecture.rules.ApiAnnotationRules;
-import org.apache.flink.architecture.rules.TableApiRules;
+package org.apache.flink.architecture.common;
 
 import com.tngtech.archunit.core.importer.ImportOption;
 import com.tngtech.archunit.core.importer.Location;
-import com.tngtech.archunit.junit.AnalyzeClasses;
-import com.tngtech.archunit.junit.ArchTest;
-import com.tngtech.archunit.junit.ArchTests;
 
 import java.util.regex.Pattern;
 
-/** Architecture tests. */
-@AnalyzeClasses(
-        packages = "org.apache.flink",
-        importOptions = {
-            ImportOption.DoNotIncludeTests.class,
-            ArchitectureTest.ExcludeScalaImportOption.class,
-            ArchitectureTest.ExcludeShadedImportOption.class
-        })
-public class ArchitectureTest {
-    @ArchTest
-    public static final ArchTests API_ANNOTATIONS = ArchTests.in(ApiAnnotationRules.class);
-
-    @ArchTest public static final ArchTests TABLE_API = ArchTests.in(TableApiRules.class);
-
-    // ---------------------------------------------------------------------------------------------
+/** Provide the most used {@link ImportOption}. */
+public class ImportOptions {
 
     /**
      * Excludes Scala classes on a best-effort basis.
@@ -55,7 +34,7 @@ public class ArchitectureTest {
      * GivenJavaClasses} or {@link SourcePredicates#areJavaClasses()} should be used in rules as
      * well.
      */
-    static class ExcludeScalaImportOption implements ImportOption {
+    public static final class ExcludeScalaImportOption implements ImportOption {
         private static final Pattern SCALA = Pattern.compile(".*/scala/.*");
 
         @Override
@@ -70,7 +49,7 @@ public boolean includes(Location location) {
      * <p>This is not only important to exclude external code shaded into a package like {@code
      * org.apache.flink.shaded.*} from being tested, but crucial for memory consumption.
      */
-    static class ExcludeShadedImportOption implements ImportOption {
+    public static final class ExcludeShadedImportOption implements ImportOption {
         private static final Pattern SHADED = Pattern.compile(".*/shaded/.*");
 
         @Override

flink-architecture-tests/flink-architecture-tests-base/src/main/java/org/apache/flink/architecture/common/JavaFieldPredicates.java

@@ -0,0 +1,104 @@
+/*
+ * 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.architecture.common;
+
+import com.tngtech.archunit.base.DescribedPredicate;
+import com.tngtech.archunit.core.domain.JavaField;
+import com.tngtech.archunit.core.domain.JavaModifier;
+
+import java.lang.annotation.Annotation;
+
+/** Fine-grained predicates focus on the JavaField. */
+public class JavaFieldPredicates {
+
+    /**
+     * Match the public modifier of the {@link JavaField}.
+     *
+     * @return A {@link DescribedPredicate} returning true, if and only if the tested {@link
+     *     JavaField} has the public modifier.
+     */
+    public static DescribedPredicate<JavaField> isPublic() {
+        return DescribedPredicate.describe(
+                "public", field -> field.getModifiers().contains(JavaModifier.PUBLIC));
+    }
+
+    /**
+     * Match the static modifier of the {@link JavaField}.
+     *
+     * @return A {@link DescribedPredicate} returning true, if and only if the tested {@link
+     *     JavaField} has the static modifier.
+     */
+    public static DescribedPredicate<JavaField> isStatic() {
+        return DescribedPredicate.describe(
+                "static", field -> field.getModifiers().contains(JavaModifier.STATIC));
+    }
+
+    /**
+     * Match none static modifier of the {@link JavaField}.
+     *
+     * @return A {@link DescribedPredicate} returning true, if and only if the tested {@link
+     *     JavaField} has no static modifier.
+     */
+    public static DescribedPredicate<JavaField> isNotStatic() {
+        return DescribedPredicate.describe(
+                "not static", field -> !field.getModifiers().contains(JavaModifier.STATIC));
+    }
+
+    /**
+     * Match the final modifier of the {@link JavaField}.
+     *
+     * @return A {@link DescribedPredicate} returning true, if and only if the tested {@link
+     *     JavaField} has the final modifier.
+     */
+    public static DescribedPredicate<JavaField> isFinal() {
+        return DescribedPredicate.describe(
+                "final", field -> field.getModifiers().contains(JavaModifier.FINAL));
+    }
+
+    /**
+     * Match the {@link Class} of the {@link JavaField}.
+     *
+     * @return A {@link DescribedPredicate} returning true, if and only if the tested {@link
+     *     JavaField} has the same type of the given {@code clazz}.
+     */
+    public static DescribedPredicate<JavaField> ofType(Class<?> clazz) {
+        return DescribedPredicate.describe(
+                "of type " + clazz.getSimpleName(),
+                field -> field.getRawType().isEquivalentTo(clazz));
+    }
+
+    /**
+     * Match the single Annotation of the {@link JavaField}.
+     *
+     * @return A {@link DescribedPredicate} returning true, if and only if the tested {@link
+     *     JavaField} has exactly the given Annotation {@code annotationType}.
+     */
+    public static DescribedPredicate<JavaField> annotatedWith(
+            Class<? extends Annotation> annotationType) {
+        return DescribedPredicate.describe(
+                "annotated with @" + annotationType.getSimpleName(),
+                field ->
+                        field.getAnnotations().size() == 1
+                                && field.getAnnotations()
+                                        .iterator()
+                                        .next()
+                                        .getRawType()
+                                        .isEquivalentTo(annotationType));
+    }
+}

flink-architecture-tests/flink-architecture-tests-base/src/main/java/org/apache/flink/architecture/common/Predicates.java

@@ -0,0 +1,149 @@
+/*
+ * 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.architecture.common;
+
+import com.tngtech.archunit.base.ChainableFunction;
+import com.tngtech.archunit.base.DescribedPredicate;
+import com.tngtech.archunit.base.Function;
+import com.tngtech.archunit.core.domain.JavaClass;
+import com.tngtech.archunit.core.domain.JavaField;
+import com.tngtech.archunit.core.domain.JavaModifier;
+import com.tngtech.archunit.core.domain.properties.CanBeAnnotated;
+
+import java.lang.annotation.Annotation;
+import java.util.Arrays;
+import java.util.Set;
+
+import static com.tngtech.archunit.lang.conditions.ArchPredicates.is;
+import static org.apache.flink.architecture.common.JavaFieldPredicates.annotatedWith;
+import static org.apache.flink.architecture.common.JavaFieldPredicates.isFinal;
+import static org.apache.flink.architecture.common.JavaFieldPredicates.isNotStatic;
+import static org.apache.flink.architecture.common.JavaFieldPredicates.isPublic;
+import static org.apache.flink.architecture.common.JavaFieldPredicates.ofType;
+
+/** Common predicates for architecture tests. */
+public class Predicates {
+
+    @SafeVarargs
+    public static DescribedPredicate<JavaClass> areDirectlyAnnotatedWithAtLeastOneOf(
+            Class<? extends Annotation>... annotations) {
+        return Arrays.stream(annotations)
+                .map(CanBeAnnotated.Predicates::annotatedWith)
+                .reduce(DescribedPredicate::or)
+                .orElseThrow(IllegalArgumentException::new)
+                .forSubtype();
+    }
+
+    /**
+     * @return A {@link DescribedPredicate} returning true, if and only if the predicate {@link
+     *     JavaField} could be found in the {@link JavaClass}.
+     */
+    public static DescribedPredicate<JavaClass> containAnyFieldsInClassHierarchyThat(
+            DescribedPredicate<? super JavaField> predicate) {
+        return new ContainAnyFieldsThatPredicate<>(
+                "fields",
+                new ChainableFunction<JavaClass, Set<JavaField>>() {
+                    @Override
+                    public Set<JavaField> apply(JavaClass input) {
+                        // need to get all fields with the inheritance hierarchy
+                        return input.getAllFields();
+                    }
+                },
+                predicate);
+    }
+
+    /**
+     * Tests that the given field is {@code public static} and of the given type {@code clazz} .
+     *
+     * <p>Attention: changing the description will add a rule into the stored.rules.
+     */
+    public static DescribedPredicate<JavaField> arePublicStaticOfType(Class<?> clazz) {
+        return DescribedPredicate.describe(
+                "are public, static, and of type " + clazz.getSimpleName(),
+                field ->
+                        field.getModifiers().contains(JavaModifier.PUBLIC)
+                                && field.getModifiers().contains(JavaModifier.STATIC)
+                                && field.getRawType().isEquivalentTo(clazz));
+    }
+
+    /**
+     * Tests that the given field is {@code public final} and not {@code static} and of the given
+     * type {@code clazz} .
+     */
+    public static DescribedPredicate<JavaField> arePublicFinalOfType(Class<?> clazz) {
+        return is(ofType(clazz)).and(isPublic()).and(isFinal()).and(isNotStatic());
+    }
+
+    /**
+     * Tests that the given field is {@code public static final} and of the given type {@code clazz}
+     * .
+     */
+    public static DescribedPredicate<JavaField> arePublicStaticFinalOfType(Class<?> clazz) {
+        return arePublicStaticOfType(clazz).and(isFinal());
+    }
+
+    /**
+     * Tests that the given field is {@code public final} and of the given type {@code clazz} with
+     * exactly the given {@code annotationType}.
+     */
+    public static DescribedPredicate<JavaField> arePublicFinalOfTypeWithAnnotation(
+            Class<?> clazz, Class<? extends Annotation> annotationType) {
+        return arePublicFinalOfType(clazz).and(annotatedWith(annotationType));
+    }
+
+    /**
+     * Tests that the given field is {@code public static final} and of the given type {@code clazz}
+     * with exactly the given {@code annotationType}.
+     */
+    public static DescribedPredicate<JavaField> arePublicStaticFinalOfTypeWithAnnotation(
+            Class<?> clazz, Class<? extends Annotation> annotationType) {
+        return arePublicStaticFinalOfType(clazz).and(annotatedWith(annotationType));
+    }
+
+    private Predicates() {}
+
+    /**
+     * A predicate to determine if a {@link JavaClass} contains one or more {@link JavaField }
+     * matching the supplied predicate.
+     */
+    private static class ContainAnyFieldsThatPredicate<T extends JavaField>
+            extends DescribedPredicate<JavaClass> {
+        private final Function<JavaClass, Set<T>> getFields;
+        private final DescribedPredicate<? super T> predicate;
+
+        ContainAnyFieldsThatPredicate(
+                String fieldDescription,
+                Function<JavaClass, Set<T>> getFields,
+                DescribedPredicate<? super T> predicate) {
+            super("contain any " + fieldDescription + " that " + predicate.getDescription());
+            this.getFields = getFields;
+            this.predicate = predicate;
+        }
+
+        @Override
+        public boolean apply(JavaClass input) {
+            for (T member : getFields.apply(input)) {
+                if (predicate.apply(member)) {
+                    return true;
+                }
+            }
+            return false;
+        }
+    }
+}