Enforce association lifecycle invariants

client/src/main/java/io/confluent/kafka/schemaregistry/client/rest/entities/requests/AssociationCreateOrUpdateInfo.java

@@ -152,24 +152,37 @@ public String toJson() throws IOException {
   }
 
   /**
-   * Applies defaults for association creation or upsert.
-   * For CREATE: schema implies frozen STRONG; frozen=false or WEAK with schema are rejected.
-   * For UPSERT: schema implies non-frozen STRONG; frozen and lifecycle are only defaulted if null.
+   * Applies defaults for an association that does not yet exist. Only call this while creating
+   * the association: an upsert that updates an existing one keeps that association's stored
+   * lifecycle and would be wrongly rejected here for carrying a schema.
+   *
+   * <p>For a create op a schema implies frozen STRONG, and frozen=false or WEAK alongside a
+   * schema are rejected. For an upsert op no schema is accepted at all, since a schema implies
+   * a topic-owned STRONG association and those can only be created together with the topic.
+   * Either way frozen and lifecycle are defaulted when null.
+   *
+   * <p>The matching check on an explicit STRONG lifecycle lives in the registry, which can
+   * exempt subjects in IMPORT mode.
+   *
+   * @param isCreate whether this is a create op rather than an upsert op
    */
   public void applyDefaults(boolean isCreate) {
+    if (!isCreate && getSchema() != null) {
+      throw new IllegalPropertyException(
+          "schema", "cannot be provided when adding an association to an existing topic; "
+              + "an association with a schema must be created with the topic");
+    }
     if (getSchema() != null) {
       if (getLifecycle() == LifecyclePolicy.WEAK) {
         throw new IllegalPropertyException(
             "lifecycle", "cannot be WEAK when schema is provided");
       }
-      if (getFrozen() != null && getFrozen() != isCreate) {
+      if (Boolean.FALSE.equals(getFrozen())) {
         throw new IllegalPropertyException(
-            "frozen", isCreate
-                ? "cannot be false when schema is provided for create"
-                : "cannot be true when creating via upsert; use create instead");
+            "frozen", "cannot be false when a schema is provided");
       }
       setLifecycle(LifecyclePolicy.STRONG);
-      setFrozen(isCreate);
+      setFrozen(true);
     } else {
       if (getFrozen() == null) {
         setFrozen(false);

client/src/main/java/io/confluent/kafka/schemaregistry/client/rest/entities/requests/AssociationCreateOrUpdateOp.java

@@ -143,24 +143,37 @@ public int hashCode() {
   }
 
   /**
-   * Applies defaults for association creation or upsert.
-   * For CREATE: schema implies frozen STRONG; frozen=false or WEAK with schema are rejected.
-   * For UPSERT: schema implies non-frozen STRONG; frozen and lifecycle are only defaulted if null.
+   * Applies defaults for an association that does not yet exist. Only call this while creating
+   * the association: an upsert that updates an existing one keeps that association's stored
+   * lifecycle and would be wrongly rejected here for carrying a schema.
+   *
+   * <p>For a create op a schema implies frozen STRONG, and frozen=false or WEAK alongside a
+   * schema are rejected. For an upsert op no schema is accepted at all, since a schema implies
+   * a topic-owned STRONG association and those can only be created together with the topic.
+   * Either way frozen and lifecycle are defaulted when null.
+   *
+   * <p>The matching check on an explicit STRONG lifecycle lives in the registry, which can
+   * exempt subjects in IMPORT mode.
+   *
+   * @param isCreate whether this is a create op rather than an upsert op
    */
   public void applyDefaults(boolean isCreate) {
+    if (!isCreate && getSchema() != null) {
+      throw new IllegalPropertyException(
+          "schema", "cannot be provided when adding an association to an existing topic; "
+              + "an association with a schema must be created with the topic");
+    }
     if (getSchema() != null) {
       if (getLifecycle() == LifecyclePolicy.WEAK) {
         throw new IllegalPropertyException(
             "lifecycle", "cannot be WEAK when schema is provided");
       }
-      if (getFrozen() != null && getFrozen() != isCreate) {
+      if (Boolean.FALSE.equals(getFrozen())) {
         throw new IllegalPropertyException(
-            "frozen", isCreate
-                ? "cannot be false when schema is provided for create"
-                : "cannot be true when creating via upsert; use create instead");
+            "frozen", "cannot be false when a schema is provided");
       }
       setLifecycle(LifecyclePolicy.STRONG);
-      setFrozen(isCreate);
+      setFrozen(true);
     } else {
       if (getFrozen() == null) {
         setFrozen(false);

client/src/main/java/io/confluent/kafka/schemaregistry/client/rest/entities/requests/AssociationCreateOrUpdateRequest.java

@@ -30,6 +30,7 @@
 import java.io.IOException;
 import java.util.List;
 import java.util.Objects;
+import java.util.stream.Collectors;
 
 @JsonInclude(JsonInclude.Include.NON_EMPTY)
 @JsonIgnoreProperties(ignoreUnknown = true)
@@ -59,12 +60,23 @@ public AssociationCreateOrUpdateRequest(
 
   public AssociationCreateOrUpdateRequest(
       AssociationOpRequest request, AssociationOp op) {
+    this(request, ImmutableList.of((AssociationCreateOrUpdateOp) op));
+  }
+
+  /**
+   * Builds a single request carrying every op in {@code ops}, so that a batch can apply them
+   * as a unit. Repeated association types are rejected when the request is applied.
+   */
+  public AssociationCreateOrUpdateRequest(
+      AssociationOpRequest request, List<? extends AssociationCreateOrUpdateOp> ops) {
     this(
         request.getResourceName(),
         request.getResourceNamespace(),
         request.getResourceId(),
         request.getResourceType(),
-        ImmutableList.of(new AssociationCreateOrUpdateInfo((AssociationCreateOrUpdateOp) op))
+        ops.stream()
+            .map(AssociationCreateOrUpdateInfo::new)
+            .collect(Collectors.toList())
     );
   }
 

client/src/main/java/io/confluent/kafka/schemaregistry/client/rest/entities/requests/AssociationOpRequest.java

@@ -28,8 +28,10 @@
 import io.confluent.kafka.schemaregistry.client.rest.exceptions.IllegalPropertyException;
 import io.confluent.kafka.schemaregistry.utils.JacksonMapper;
 import java.io.IOException;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Objects;
+import java.util.Set;
 
 @JsonInclude(JsonInclude.Include.NON_EMPTY)
 @JsonIgnoreProperties(ignoreUnknown = true)
@@ -141,10 +143,20 @@
     return JacksonMapper.INSTANCE.writeValueAsString(this);
   }
 
+  private static String associationTypeOf(AssociationOp op) {
+    if (op instanceof AssociationCreateOrUpdateOp) {
+      return ((AssociationCreateOrUpdateOp) op).getAssociationType();
+    }
+    if (op instanceof AssociationDeleteOp) {
+      return ((AssociationDeleteOp) op).getAssociationType();
+    }
+    return null;
+  }
+
   // Validates resource fields, then for each op: validates the op, defaults
   // the subject to :.ns:name-type for STRONG associations, and enforces that
   // frozen associations use the default subject.
   public void validate(boolean dryRun) {
     checkName(getResourceName(), "resourceName");
     checkName(getResourceNamespace(), "resourceNamespace");
     if (!dryRun && (getResourceId() == null || getResourceId().isEmpty())) {
@@ -161,8 +173,16 @@
     if (getAssociations() == null || getAssociations().isEmpty()) {
       throw new IllegalPropertyException("associations", "cannot be null or empty");
     }
+    Set<String> associationTypes = new HashSet<>();
     for (AssociationOp op : getAssociations()) {
       op.validate(dryRun);
+      String associationType = associationTypeOf(op);
+      if (associationType != null && !associationTypes.add(associationType)) {
+        throw new IllegalPropertyException(
+            "associationType",
+            "may only appear once per request, but '" + associationType + "' appears more"
+                + " than once");
+      }
       if (op instanceof AssociationCreateOrUpdateOp) {
         AssociationCreateOrUpdateOp createOrUpdateOp = (AssociationCreateOrUpdateOp) op;
         if (op instanceof AssociationCreateOp) {

client/src/test/java/io/confluent/kafka/schemaregistry/client/rest/entities/requests/AssociationsRequestTest.java

@@ -32,7 +32,7 @@
   @Test
   public void testInfoValidLifecycleDoesNotThrow() {
     AssociationCreateOrUpdateInfo info = new AssociationCreateOrUpdateInfo(
         "test-subject", null, LifecyclePolicy.STRONG, null, null, null);
     info.validate(false, false);
   }
 
@@ -100,7 +100,7 @@
   @Test
   public void testCreateOpWithSchemaDefaultsToStrongFrozen() {
     RegisterSchemaRequest schema = new RegisterSchemaRequest();
     schema.setSchema("{\"type\":\"string\"}");
     AssociationCreateOp op = new AssociationCreateOp(
         "test-subject", null, null, null, schema, null);
     op.validate(false);
@@ -213,6 +213,105 @@
     info.validate(false, false);
   }
 
+  // An upsert may only create a WEAK association: a STRONG association is owned by its topic
+  // and has to be created with it, and a schema implies STRONG.
+
+  @Test(expected = IllegalPropertyException.class)
+  public void testUpsertInfoCreatingWithSchemaAndSubjectThrows() {
+    RegisterSchemaRequest schema = new RegisterSchemaRequest();
+    schema.setSchema("{\"type\":\"string\"}");
+    AssociationCreateOrUpdateInfo info = new AssociationCreateOrUpdateInfo(
+        "test-subject", null, null, null, schema, null);
+    info.applyDefaults(false);
+  }
+
+  @Test(expected = IllegalPropertyException.class)
+  public void testUpsertInfoCreatingWithSchemaAndStrongLifecycleThrows() {
+    RegisterSchemaRequest schema = new RegisterSchemaRequest();
+    schema.setSchema("{\"type\":\"string\"}");
+    AssociationCreateOrUpdateInfo info = new AssociationCreateOrUpdateInfo(
+        "test-subject", null, LifecyclePolicy.STRONG, null, schema, null);
+    info.applyDefaults(false);
+  }
+
+  @Test(expected = IllegalPropertyException.class)
+  public void testUpsertInfoCreatingWithSchemaAndNoSubjectThrows() {
+    RegisterSchemaRequest schema = new RegisterSchemaRequest();
+    schema.setSchema("{\"type\":\"string\"}");
+    AssociationCreateOrUpdateInfo info = new AssociationCreateOrUpdateInfo(
+        null, null, null, null, schema, null);
+    info.applyDefaults(false);
+  }
+
+  /**
+   * An explicit STRONG lifecycle without a schema is left alone here: the registry rejects it,
+   * so that subjects in IMPORT mode can be exempted.
+   */
+  @Test
+  public void testUpsertInfoCreatingWithStrongLifecycleAndNoSchemaIsLeftToRegistry() {
+    AssociationCreateOrUpdateInfo info = new AssociationCreateOrUpdateInfo(
+        "test-subject", null, LifecyclePolicy.STRONG, null, null, null);
+    info.applyDefaults(false);
+    assertEquals(LifecyclePolicy.STRONG, info.getLifecycle());
+    assertEquals(Boolean.FALSE, info.getFrozen());
+  }
+
+  @Test
+  public void testUpsertInfoCreatingFrozenStrongKeepsFrozenForImport() {
+    AssociationCreateOrUpdateInfo info = new AssociationCreateOrUpdateInfo(
+        "test-subject", null, LifecyclePolicy.STRONG, true, null, null);
+    info.applyDefaults(false);
+    assertEquals(LifecyclePolicy.STRONG, info.getLifecycle());
+    assertEquals(Boolean.TRUE, info.getFrozen());
+  }
+
+  @Test
+  public void testUpsertInfoCreatingWeakDoesNotThrow() {
+    AssociationCreateOrUpdateInfo info = new AssociationCreateOrUpdateInfo(
+        "test-subject", null, null, null, null, null);
+    info.applyDefaults(false);
+    assertEquals(LifecyclePolicy.WEAK, info.getLifecycle());
+    assertEquals(Boolean.FALSE, info.getFrozen());
+  }
+
+  @Test
+  public void testCreateInfoWithSchemaAndSubjectStillDefaultsToStrongFrozen() {
+    RegisterSchemaRequest schema = new RegisterSchemaRequest();
+    schema.setSchema("{\"type\":\"string\"}");
+    AssociationCreateOrUpdateInfo info = new AssociationCreateOrUpdateInfo(
+        "test-subject", null, null, null, schema, null);
+    info.applyDefaults(true);
+    assertEquals(LifecyclePolicy.STRONG, info.getLifecycle());
+    assertEquals(Boolean.TRUE, info.getFrozen());
+  }
+
+  @Test(expected = IllegalPropertyException.class)
+  public void testUpsertOpCreatingWithSchemaAndSubjectThrows() {
+    RegisterSchemaRequest schema = new RegisterSchemaRequest();
+    schema.setSchema("{\"type\":\"string\"}");
+    AssociationUpsertOp op = new AssociationUpsertOp(
+        "test-subject", null, null, null, schema, null);
+    op.applyDefaults(false);
+  }
+
+  @Test(expected = IllegalPropertyException.class)
+  public void testUpsertOpCreatingWithSchemaAndNoSubjectThrows() {
+    RegisterSchemaRequest schema = new RegisterSchemaRequest();
+    schema.setSchema("{\"type\":\"string\"}");
+    AssociationUpsertOp op = new AssociationUpsertOp(
+        null, null, null, null, schema, null);
+    op.applyDefaults(false);
+  }
+
+  @Test
+  public void testUpsertOpCreatingWithStrongLifecycleAndNoSchemaIsLeftToRegistry() {
+    AssociationUpsertOp op = new AssociationUpsertOp(
+        "test-subject", null, LifecyclePolicy.STRONG, null, null, null);
+    op.applyDefaults(false);
+    assertEquals(LifecyclePolicy.STRONG, op.getLifecycle());
+    assertEquals(Boolean.FALSE, op.getFrozen());
+  }
+
   // Requirement #3: Default subject + subject required
 
   @Test