[feat][misc] PIP-478: dedicated pulsar-tls-factory-api and pulsar-http-client-api SPI modules

build-logic/conventions/src/main/kotlin/CheckBinaryLicenseTask.kt

@@ -65,7 +65,9 @@ abstract class CheckBinaryLicenseTask : DefaultTask() {
             "pulsar-client",
             "pulsar-cli-utils",
             "pulsar-common",
+            "pulsar-http-client-api",
             "pulsar-package",
+            "pulsar-tls-factory-api",
             "pulsar-websocket",
         )
 

pulsar-bom/build.gradle.kts

@@ -37,6 +37,10 @@ dependencies {
         api(project(":pulsar-client-api"))
         api(project(":pulsar-client-admin-api"))
 
+        // Focused SPI modules (PIP-478): TLS factory SPI + HTTP client SPI
+        api(project(":pulsar-tls-factory-api"))
+        api(project(":pulsar-http-client-api"))
+
         // Shaded clients (the published artifacts users depend on)
         api(project(":pulsar-client-shaded"))
         api(project(":pulsar-client-admin-shaded"))

pulsar-common/build.gradle.kts

@@ -145,6 +145,11 @@ sourceSets["main"].resources.srcDir(generatePulsarBuildInfo.map {
 
 dependencies {
     implementation(libs.slog)
+    // PIP-478: the purpose-driven TLS factory SPI (org.apache.pulsar.tls) lives in the focused
+    // pulsar-tls-factory-api module; the default FileBasedTlsFactory impl (org.apache.pulsar.common.tls.impl)
+    // and the hostname-verification helpers (org.apache.pulsar.common.tls) live here. Exposed as `api` so
+    // consumers that reference the SPI through pulsar-common keep compiling unchanged.
+    api(project(":pulsar-tls-factory-api"))
     api(project(":pulsar-client-api"))
     api(project(":pulsar-client-admin-api"))
 
@@ -196,6 +201,11 @@ dependencies {
 
     compileOnly(libs.swagger.annotations)
     compileOnly(libs.spotbugs.annotations)
+    // PIP-478: FileBasedTlsFactory emits the pulsar.tls.* reload instruments via the OpenTelemetry handle
+    // exposed on TlsFactoryInitContext. Kept compileOnly (matching pulsar-tls-factory-api) — the real
+    // OpenTelemetry root is always supplied at runtime by the owning component (broker/client), and a noop
+    // root yields no-op instruments.
+    compileOnly(libs.opentelemetry.api)
 
     // Non-FIPS BouncyCastle provider for tests that exercise SecurityUtility (which loads
     // org.bouncycastle.jce.provider.BouncyCastleProvider in a static initializer). This matches
@@ -208,4 +218,10 @@ dependencies {
     testImplementation(libs.snappy.java)
     testImplementation(libs.awaitility)
     testImplementation(libs.jsonassert)
+    // PIP-478: the TLS factory tests implement TlsFactoryInitContext, whose openTelemetry() accessor
+    // exposes the (compileOnly) OpenTelemetry API from pulsar-tls-factory-api.
+    testImplementation(libs.opentelemetry.api)
+    // PIP-478: the TLS-reload metrics test reads pulsar.tls.* instruments via an in-memory SDK reader.
+    testImplementation(libs.opentelemetry.sdk)
+    testImplementation(libs.opentelemetry.sdk.testing)
 }

pulsar-common/src/test/java/org/apache/pulsar/tls/TlsPolicyValidationTest.java

@@ -0,0 +1,76 @@
+/*
+ * 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.pulsar.tls;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import org.testng.annotations.Test;
+
+/**
+ * PIP-478: {@link TlsPolicy.Builder#build()} fails loud when a configured field is inconsistent with the
+ * chosen {@link TlsPolicy.Format}, rather than silently ignoring it.
+ */
+public class TlsPolicyValidationTest {
+
+    @Test
+    public void pemPolicyRejectsKeystoreFields() {
+        assertThatThrownBy(() -> TlsPolicy.builder()
+                .format(TlsPolicy.Format.PEM)
+                .trustCertsFilePath("/ca.pem")
+                .keyStorePath("/key.p12")
+                .build())
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining("keyStorePath");
+
+        assertThatThrownBy(() -> TlsPolicy.builder()
+                .format(TlsPolicy.Format.PEM)
+                .keyStoreType("PKCS12")
+                .build())
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining("keyStoreType");
+    }
+
+    @Test
+    public void keystorePolicyRejectsPemFields() {
+        assertThatThrownBy(() -> TlsPolicy.builder()
+                .format(TlsPolicy.Format.KEYSTORE)
+                .keyStorePath("/key.p12")
+                .certificateFilePath("/cert.pem")
+                .build())
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining("certificateFilePath");
+    }
+
+    @Test
+    public void consistentPoliciesBuild() {
+        assertThatCode(() -> TlsPolicy.pem("/ca.pem", "/cert.pem", "/key.pem"))
+                .doesNotThrowAnyException();
+        assertThatCode(() -> TlsPolicy.keyStore("/trust.jks", "pw", "/key.jks", "pw", "JKS"))
+                .doesNotThrowAnyException();
+        // Mixed store TYPES are consistent with KEYSTORE format (only cross-format fields are rejected).
+        assertThatCode(() -> TlsPolicy.builder().format(TlsPolicy.Format.KEYSTORE)
+                .keyStorePath("/key.p12").keyStoreType("PKCS12")
+                .trustStorePath("/trust.jks").trustStoreType("JKS")
+                .build()).doesNotThrowAnyException();
+        // Flag-only policies (system default / insecure) build cleanly.
+        assertThat(TlsPolicy.builder().build().format()).isEqualTo(TlsPolicy.Format.PEM);
+        assertThat(TlsPolicy.insecure().allowInsecureConnection()).isTrue();
+    }
+}

pulsar-common/src/test/java/org/apache/pulsar/tls/TlsPurposeTest.java

@@ -0,0 +1,59 @@
+/*
+ * 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.pulsar.tls;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import java.util.HashMap;
+import java.util.Map;
+import org.testng.annotations.Test;
+
+/**
+ * Locks the {@link TlsPurpose} identity contract (PIP-478): equality is over {@code (role, name)}, so a
+ * purpose resolves to the same purpose→policy map slot regardless of how it was minted.
+ */
+public class TlsPurposeTest {
+
+    @Test
+    public void equalsAndHashCodeOverRoleAndName() {
+        assertThat(TlsPurpose.client("x")).isEqualTo(TlsPurpose.client("x"));
+        assertThat(TlsPurpose.client("x").hashCode()).isEqualTo(TlsPurpose.client("x").hashCode());
+    }
+
+    @Test
+    public void purposeIsAStableMapKey() {
+        Map<TlsPurpose, String> map = new HashMap<>();
+        map.put(TlsPurpose.client("x"), "policy");
+
+        assertThat(map).containsKey(TlsPurpose.client("x"));
+        assertThat(map.get(TlsPurpose.client("x"))).isEqualTo("policy");
+    }
+
+    @Test
+    public void roleAndNameDistinguish() {
+        assertThat(TlsPurpose.client("x")).isNotEqualTo(TlsPurpose.server("x"));
+        assertThat(TlsPurpose.client("x")).isNotEqualTo(TlsPurpose.client("y"));
+    }
+
+    @Test
+    public void mintedPurposeCarriesRoleAndName() {
+        TlsPurpose server = TlsPurpose.server("broker.internal");
+        assertThat(server.role()).isEqualTo(TlsPurpose.Role.SERVER);
+        assertThat(server.name()).isEqualTo("broker.internal");
+    }
+}

pulsar-http-client-api/build.gradle.kts

@@ -0,0 +1,31 @@
+/*
+ * 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.
+ */
+
+// PIP-478: focused, dependency-light API module hosting the framework-managed HTTP client SPI
+// (org.apache.pulsar.http). Published so that server-side modules and the v5 client may depend on it
+// in later stages — the published-module dependency guard only allows published-on-published deps.
+plugins {
+    id("pulsar.public-java-library-conventions")
+}
+
+dependencies {
+    // PulsarHttpClientConfig selects TLS material by TlsPurpose, which is part of this SPI's surface;
+    // exposed as `api` so consumers of PulsarHttpClientConfig see the TLS factory SPI.
+    api(project(":pulsar-tls-factory-api"))
+}

pulsar-http-client-api/src/main/java/org/apache/pulsar/http/HttpRequest.java

@@ -0,0 +1,161 @@
+/*
+ * 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.pulsar.http;
+
+import java.net.URI;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Optional;
+
+/**
+ * An immutable HTTP request (PIP-478).
+ *
+ * <p>Construct via {@link #builder(Method, URI)}. The optional {@link Body} is a sealed type so the
+ * set of body shapes (currently only {@link Bytes}) is closed; the framework's single body-encoding
+ * site dispatches on the body type.
+ *
+ * <p><b>Headers are single-valued and keyed by exact name.</b> Unlike {@link HttpResponse} (whose names
+ * are canonicalised), request header names are NOT case-folded: they are stored verbatim as supplied to
+ * the builder. Setting a header whose name exactly matches an existing one replaces the prior value
+ * (last-wins); names that differ only in case are therefore distinct entries, and multi-valued headers are
+ * not representable.
+ */
+public final class HttpRequest {
+
+    /** The HTTP method. */
+    public enum Method {
+        GET, POST, PUT, DELETE, HEAD, OPTIONS, PATCH
+    }
+
+    /** A request body. Sealed to the single {@link Bytes} shape. */
+    public sealed interface Body permits Bytes {
+    }
+
+    /**
+     * A raw byte body with an explicit content type.
+     *
+     * <p><b>Ownership: the array is handed off, not copied.</b> The caller transfers ownership of
+     * {@code content} to this body on construction and the framework reads it without copying; neither
+     * side may mutate the array after construction. Because the component is an array, the record's
+     * generated {@code equals}/{@code hashCode} are reference-based, not value-based; do not compare
+     * {@link Bytes} instances for value equality.
+     *
+     * @param content     the body bytes
+     * @param contentType the {@code Content-Type} value
+     */
+    public record Bytes(byte[] content, String contentType) implements Body {
+    }
+
+    private final Method method;
+    private final URI uri;
+    private final Map<String, String> headers;
+    private final Body body;
+
+    private HttpRequest(Builder b) {
+        this.method = b.method;
+        this.uri = b.uri;
+        this.headers = Collections.unmodifiableMap(new LinkedHashMap<>(b.headers));
+        this.body = b.body;
+    }
+
+    /**
+     * @return the HTTP method
+     */
+    public Method method() {
+        return method;
+    }
+
+    /**
+     * @return the request URI
+     */
+    public URI uri() {
+        return uri;
+    }
+
+    /**
+     * @return an unmodifiable view of the request headers
+     */
+    public Map<String, String> headers() {
+        return headers;
+    }
+
+    /**
+     * @return the request body, if any
+     */
+    public Optional<Body> body() {
+        return Optional.ofNullable(body);
+    }
+
+    /**
+     * Create a builder.
+     *
+     * @param method the HTTP method
+     * @param uri    the request URI
+     * @return a new {@link Builder}
+     */
+    public static Builder builder(Method method, URI uri) {
+        return new Builder(method, uri);
+    }
+
+    /**
+     * Builder for {@link HttpRequest}.
+     */
+    public static final class Builder {
+        private final Method method;
+        private final URI uri;
+        private final Map<String, String> headers = new LinkedHashMap<>();
+        private Body body;
+
+        private Builder(Method method, URI uri) {
+            this.method = method;
+            this.uri = uri;
+        }
+
+        /**
+         * Add or replace a header.
+         *
+         * @param name  the header name
+         * @param value the header value
+         * @return this builder
+         */
+        public Builder header(String name, String value) {
+            headers.put(name, value);
+            return this;
+        }
+
+        /**
+         * Set the request body.
+         *
+         * @param body the body
+         * @return this builder
+         */
+        public Builder body(Body body) {
+            this.body = body;
+            return this;
+        }
+
+        /**
+         * @return a new immutable {@link HttpRequest}
+         */
+        public HttpRequest build() {
+            return new HttpRequest(this);
+        }
+    }
+}