[improve][schema] Add admin cli for testCompatibility (#19974)

<!--
### Contribution Checklist
  
  - PR title format should be *[type][component] summary*. For details, see *[Guideline - Pulsar PR Naming Convention](https://pulsar.apache.org/contribute/develop-semantic-title/)*. 

  - Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
  
  - Each pull request should address only one issue, not mix up code from multiple issues.
  
  - Each commit in the pull request has a meaningful commit message

  - Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
-->

<!-- Either this PR fixes an issue, -->

<!-- or this PR is one task of an issue -->

<!-- If the PR belongs to a PIP, please add the PIP link here -->

<!-- Details of when a PIP is required and how the PIP process work, please see: https://github.com/apache/pulsar/blob/master/wiki/proposals/PIP.md -->

### Motivation

1. add admin cli `testCompatibility`
2. add `admin.schemas().testCompatibility()` test
### Modifications

1. add admin cli `testCompatibility`
2. add `admin.schemas().testCompatibility()` test
### Verifying this change
add test for it
- [ ] Make sure that the change passes the CI checks.

*(Please pick either of the following options)*

This change is a trivial rework / code cleanup without any test coverage.

*(or)*

This change is already covered by existing tests, such as *(please describe tests)*.

*(or)*

This change added tests and can be verified as follows:

*(example:)*
  - *Added integration tests for end-to-end deployment with large payloads (10MB)*
  - *Extended integration test for recovery after broker failure*

### Does this pull request potentially affect one of the following parts:

<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->

*If the box was checked, please highlight the changes*

- [ ] Dependencies (add or upgrade a dependency)
- [ ] The public API
- [ ] The schema
- [ ] The default values of configurations
- [ ] The threading model
- [ ] The binary protocol
- [ ] The REST endpoints
- [x] The admin CLI options
- [ ] The metrics
- [ ] Anything that affects deployment

### Documentation

<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->

- [ ] `doc` <!-- Your PR contains doc changes. -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->

### Matching PR in forked repository

PR in forked repository: <!-- ENTER URL HERE -->
congbobo184#14

<!--
After opening this PR, the build in apache/pulsar will fail and instructions will
be provided for opening a PR in the PR author's forked repository.

apache/pulsar pull requests should be first tested in your own fork since the 
apache/pulsar CI based on GitHub Actions has constrained resources and quota.
GitHub Actions provides separate quota for pull requests that are executed in 
a forked repository.

The tests will be run in the forked repository until all PR review comments have
been handled, the tests pass and the PR is approved by a reviewer.
-->

Author: congbobo184

pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiSchemaTest.java

@@ -49,6 +49,8 @@
 import org.apache.pulsar.common.policies.data.SchemaAutoUpdateCompatibilityStrategy;
 import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy;
 import org.apache.pulsar.common.policies.data.TenantInfoImpl;
+import org.apache.pulsar.common.protocol.schema.IsCompatibilityResponse;
+import org.apache.pulsar.common.protocol.schema.PostSchemaPayload;
 import org.apache.pulsar.common.schema.SchemaInfo;
 import org.apache.pulsar.common.schema.SchemaInfoWithVersion;
 import org.apache.pulsar.common.schema.SchemaType;
@@ -438,4 +440,31 @@ public void testGetSchemaCompatibilityStrategyWhenSetBrokerLevelAndSchemaAutoUpd
                 admin.namespaces().getSchemaCompatibilityStrategy(schemaCompatibilityNamespace),
                 SchemaCompatibilityStrategy.UNDEFINED));
     }
+
+    @Test
+    public void testCompatibility() throws Exception {
+        String topicName = schemaCompatibilityNamespace + "/testCompatibility";
+        try {
+            admin.schemas().getSchemaInfo(topicName);
+            fail();
+        } catch (PulsarAdminException.NotFoundException e) {
+            assertEquals(e.getMessage(), "Schema not found");
+        }
+        Map<String, String> properties = new HashMap<>();
+        PostSchemaPayload postSchemaPayload = new PostSchemaPayload("STRING", "", properties);
+        admin.schemas().createSchema(topicName, postSchemaPayload);
+        IsCompatibilityResponse isCompatibilityResponse =
+                admin.schemas().testCompatibility(topicName, postSchemaPayload);
+
+        assertTrue(isCompatibilityResponse.isCompatibility());
+        assertEquals(isCompatibilityResponse.getSchemaCompatibilityStrategy(), SchemaCompatibilityStrategy.FULL.name());
+        postSchemaPayload = new PostSchemaPayload("INT8", "", properties);
+        try {
+            admin.schemas().testCompatibility(topicName, postSchemaPayload);
+            fail();
+        } catch (Exception e) {
+            assertTrue(e instanceof PulsarAdminException.ServerSideErrorException);
+            assertTrue(e.getMessage().contains("Incompatible schema: exists schema type STRING, new schema type INT8"));
+        }
+    }
 }

pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/protocol/schema/IsCompatibilityResponse.java

@@ -33,5 +33,4 @@
 public class IsCompatibilityResponse {
     boolean isCompatibility;
     String schemaCompatibilityStrategy;
-
 }

pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/PulsarAdminToolTest.java

@@ -2360,6 +2360,11 @@ void schemas() throws Exception {
         PostSchemaPayload input = new ObjectMapper().readValue(new File(schemaFile), PostSchemaPayload.class);
         verify(schemas).createSchema("persistent://tn1/ns1/tp1", input);
 
+        cmdSchemas = new CmdSchemas(() -> admin);
+        cmdSchemas.run(split("compatibility -f " + schemaFile + " persistent://tn1/ns1/tp1"));
+        input = new ObjectMapper().readValue(new File(schemaFile), PostSchemaPayload.class);
+        verify(schemas).testCompatibility("persistent://tn1/ns1/tp1", input);
+
         cmdSchemas = new CmdSchemas(() -> admin);
         String jarFile = PulsarAdminToolTest.class.getClassLoader()
                 .getResource("dummyexamples.jar").getFile();

pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdSchemas.java

@@ -42,6 +42,7 @@ public CmdSchemas(Supplier<PulsarAdmin> admin) {
         jcommander.addCommand("delete", new DeleteSchema());
         jcommander.addCommand("upload", new UploadSchema());
         jcommander.addCommand("extract", new ExtractSchema());
+        jcommander.addCommand("compatibility", new TestCompatibility());
     }
 
     @Parameters(commandDescription = "Get the schema for a topic")
@@ -164,4 +165,20 @@ void run() throws Exception {
         }
     }
 
+    @Parameters(commandDescription = "Test schema compatibility")
+    private class TestCompatibility extends CliCommand {
+        @Parameter(description = "persistent://tenant/namespace/topic", required = true)
+        private java.util.List<String> params;
+
+        @Parameter(names = { "-f", "--filename" }, description = "filename", required = true)
+        private String schemaFileName;
+
+        @Override
+        void run() throws Exception {
+            String topic = validateTopicName(params);
+            PostSchemaPayload input = MAPPER.readValue(new File(schemaFileName), PostSchemaPayload.class);
+            getAdmin().schemas().testCompatibility(topic, input);
+        }
+    }
+
 }