merge changes from 5.x.x to main

.github/workflows/integration-test.yml

@@ -94,8 +94,9 @@ jobs:
           EOF
       # Every job gets its own Cloud Foundry on KinD, so the matrix can run in parallel.
       - name: Set up Cloud Foundry
-        uses: cloudfoundry/kind-deployment/.github/actions/setup-cf@main
+        uses: cloudfoundry/kind-deployment/.github/actions/setup-cf@b42a88a35756595bd4108d2b525ded5b014a9515 # v0.26.0
         with:
+          ref: b42a88a35756595bd4108d2b525ded5b014a9515
           use-latest-versions: ${{ (inputs.preview || github.event_name == 'schedule') }}
           github-token: ${{ github.token }}
           additional-values-files: ${{ runner.temp }}/cf-values.yaml

cloudfoundry-client-reactor/src/main/java/org/cloudfoundry/reactor/client/_ReactorCloudFoundryClient.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2021 the original author or authors.
+ * Copyright 2013-2026 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -58,6 +58,7 @@
 import org.cloudfoundry.client.v3.deployments.DeploymentsV3;
 import org.cloudfoundry.client.v3.domains.DomainsV3;
 import org.cloudfoundry.client.v3.droplets.Droplets;
+import org.cloudfoundry.client.v3.featureflags.FeatureFlagsV3;
 import org.cloudfoundry.client.v3.isolationsegments.IsolationSegments;
 import org.cloudfoundry.client.v3.jobs.JobsV3;
 import org.cloudfoundry.client.v3.quotas.organizations.OrganizationQuotasV3;
@@ -120,6 +121,7 @@
 import org.cloudfoundry.reactor.client.v3.deployments.ReactorDeploymentsV3;
 import org.cloudfoundry.reactor.client.v3.domains.ReactorDomainsV3;
 import org.cloudfoundry.reactor.client.v3.droplets.ReactorDroplets;
+import org.cloudfoundry.reactor.client.v3.featureflags.ReactorFeatureFlagsV3;
 import org.cloudfoundry.reactor.client.v3.isolationsegments.ReactorIsolationSegments;
 import org.cloudfoundry.reactor.client.v3.jobs.ReactorJobsV3;
 import org.cloudfoundry.reactor.client.v3.quotas.organizations.ReactorOrganizationQuotasV3;
@@ -256,6 +258,12 @@ public FeatureFlags featureFlags() {
         return new ReactorFeatureFlags(getConnectionContext(), getRootV2(), getTokenProvider(), getRequestTags());
     }
 
+    @Override
+    @Value.Derived
+    public FeatureFlagsV3 featureFlagsV3() {
+        return new ReactorFeatureFlagsV3(getConnectionContext(), getRootV3(), getTokenProvider(), getRequestTags());
+    }
+
     @Override
     @Value.Derived
     public Info info() {

cloudfoundry-client-reactor/src/main/java/org/cloudfoundry/reactor/client/v3/featureflags/ReactorFeatureFlagsV3.java

@@ -0,0 +1,80 @@
+/*
+ * Copyright 2026 the original author or authors.
+ *
+ * Licensed 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.cloudfoundry.reactor.client.v3.featureflags;
+
+import java.util.Map;
+import org.cloudfoundry.client.v3.featureflags.FeatureFlagsV3;
+import org.cloudfoundry.client.v3.featureflags.GetFeatureFlagRequest;
+import org.cloudfoundry.client.v3.featureflags.GetFeatureFlagResponse;
+import org.cloudfoundry.client.v3.featureflags.ListFeatureFlagsRequest;
+import org.cloudfoundry.client.v3.featureflags.ListFeatureFlagsResponse;
+import org.cloudfoundry.client.v3.featureflags.UpdateFeatureFlagRequest;
+import org.cloudfoundry.client.v3.featureflags.UpdateFeatureFlagResponse;
+import org.cloudfoundry.reactor.ConnectionContext;
+import org.cloudfoundry.reactor.TokenProvider;
+import org.cloudfoundry.reactor.client.v3.AbstractClientV3Operations;
+import reactor.core.publisher.Mono;
+
+/**
+ * The Reactor-based implementation of {@link FeatureFlagsV3}
+ */
+public final class ReactorFeatureFlagsV3 extends AbstractClientV3Operations
+        implements FeatureFlagsV3 {
+
+    /**
+     * Creates an instance
+     *
+     * @param connectionContext the {@link ConnectionContext} to use when communicating with the server
+     * @param root              the root URI of the server. Typically something like {@code https://api.run.pivotal.io}.
+     * @param tokenProvider     the {@link TokenProvider} to use when communicating with the server
+     * @param requestTags       map with custom http headers which will be added to web request
+     */
+    public ReactorFeatureFlagsV3(
+            ConnectionContext connectionContext,
+            Mono<String> root,
+            TokenProvider tokenProvider,
+            Map<String, String> requestTags) {
+        super(connectionContext, root, tokenProvider, requestTags);
+    }
+
+    @Override
+    public Mono<GetFeatureFlagResponse> get(GetFeatureFlagRequest request) {
+        return get(
+                        request,
+                        GetFeatureFlagResponse.class,
+                        builder -> builder.pathSegment("feature_flags", request.getName()))
+                .checkpoint();
+    }
+
+    @Override
+    public Mono<ListFeatureFlagsResponse> list(ListFeatureFlagsRequest request) {
+        return get(
+                        request,
+                        ListFeatureFlagsResponse.class,
+                        builder -> builder.pathSegment("feature_flags"))
+                .checkpoint();
+    }
+
+    @Override
+    public Mono<UpdateFeatureFlagResponse> update(UpdateFeatureFlagRequest request) {
+        return patch(
+                        request,
+                        UpdateFeatureFlagResponse.class,
+                        builder -> builder.pathSegment("feature_flags", request.getName()))
+                .checkpoint();
+    }
+}

cloudfoundry-client-reactor/src/main/java/org/cloudfoundry/reactor/client/v3/packages/ReactorPackages.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2021 the original author or authors.
+ * Copyright 2013-2026 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -131,7 +131,9 @@ public Mono<ListPackageDropletsResponse> listDroplets(ListPackageDropletsRequest
     public Mono<UploadPackageResponse> upload(UploadPackageRequest request) {
         Path bits = request.getBits();
 
-        if (bits.toFile().isDirectory()) {
+        // bits is null when uploading via resource matching only (all files already cached on
+        // server)
+        if (bits != null && bits.toFile().isDirectory()) {
             return FileUtils.compress(bits)
                     .map(
                             temporaryFile ->

cloudfoundry-client-reactor/src/main/java/org/cloudfoundry/reactor/logcache/v1/ReactorLogCacheEndpoints.java

@@ -16,19 +16,39 @@
 
 package org.cloudfoundry.reactor.logcache.v1;
 
+import java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
 import java.util.Map;
+import org.cloudfoundry.AbstractCloudFoundryException;
+import org.cloudfoundry.logcache.v1.Envelope;
+import org.cloudfoundry.logcache.v1.EnvelopeType;
 import org.cloudfoundry.logcache.v1.InfoRequest;
 import org.cloudfoundry.logcache.v1.InfoResponse;
 import org.cloudfoundry.logcache.v1.MetaRequest;
 import org.cloudfoundry.logcache.v1.MetaResponse;
 import org.cloudfoundry.logcache.v1.ReadRequest;
 import org.cloudfoundry.logcache.v1.ReadResponse;
+import org.cloudfoundry.logcache.v1.TailLogsRequest;
 import org.cloudfoundry.reactor.ConnectionContext;
 import org.cloudfoundry.reactor.TokenProvider;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.util.retry.Retry;
 
 final class ReactorLogCacheEndpoints extends AbstractLogCacheOperations {
 
+    private static final Logger LOGGER = LoggerFactory.getLogger(ReactorLogCacheEndpoints.class);
+
+    private static final int MAX_RETRIES = 5;
+
+    private static final Duration MAX_BACKOFF = Duration.ofSeconds(10);
+
     ReactorLogCacheEndpoints(
             ConnectionContext connectionContext,
             Mono<String> root,
@@ -48,4 +68,116 @@ Mono<MetaResponse> meta(MetaRequest request) {
     Mono<ReadResponse> read(ReadRequest request) {
         return get(request, ReadResponse.class, "read", request.getSourceId()).checkpoint();
     }
+
+    /**
+     * Continuously polls Log Cache and emits new {@link Envelope}s as they arrive.
+     *
+     * <p>Mirrors the Go {@code logcache.Walk()} / {@code WalkWithDelay} semantics:
+     * <ol>
+     *   <li>On every subscription, start a cursor at {@code startTime} (defaults to now minus 5 s
+     *       in nanoseconds). The cursor is per subscription, so subscribers never interfere.</li>
+     *   <li>Issue {@code GET /api/v1/read/{sourceId}?start_time=cursor}.</li>
+     *   <li>Log Cache is eventually consistent, so envelopes younger than {@code delay} are
+     *       withheld and picked up by a later poll. The rest are emitted in ascending timestamp
+     *       order and the cursor advances to {@code lastTimestamp + 1}.</li>
+     *   <li>When nothing can be emitted, wait {@code pollInterval} before polling again.</li>
+     *   <li>Transient failures (I/O errors, HTTP 429 and 5xx) are retried with exponential
+     *       backoff, up to {@value #MAX_RETRIES} times in a row; other errors terminate the
+     *       stream.</li>
+     * </ol>
+     * The stream never completes on its own – the caller cancels the subscription to stop.
+     */
+    Flux<Envelope> logsTail(TailLogsRequest request) {
+        List<EnvelopeType> envelopeTypes =
+                request.getEnvelopeTypes() != null
+                        ? request.getEnvelopeTypes()
+                        : Collections.emptyList();
+
+        return Mono.fromSupplier(
+                        () ->
+                                request.getStartTime() != null
+                                        ? request.getStartTime()
+                                        : toNanos(Instant.now().minusSeconds(5)))
+                .flatMap(cursor -> poll(request, envelopeTypes, cursor))
+                .expand(
+                        page ->
+                                page.envelopes.isEmpty()
+                                        ? Mono.delay(request.getPollInterval())
+                                                .then(poll(request, envelopeTypes, page.cursor))
+                                        : poll(request, envelopeTypes, page.nextCursor()))
+                .concatMapIterable(page -> page.envelopes);
+    }
+
+    private Mono<Page> poll(
+            TailLogsRequest request, List<EnvelopeType> envelopeTypes, long cursor) {
+        ReadRequest.Builder builder =
+                ReadRequest.builder().sourceId(request.getSourceId()).startTime(cursor);
+        if (!envelopeTypes.isEmpty()) {
+            builder.envelopeTypes(envelopeTypes);
+        }
+        if (request.getNameFilter() != null && !request.getNameFilter().isEmpty()) {
+            builder.nameFilter(request.getNameFilter());
+        }
+
+        return read(builder.build())
+                .retryWhen(
+                        Retry.backoff(MAX_RETRIES, request.getPollInterval())
+                                .maxBackoff(MAX_BACKOFF)
+                                .filter(ReactorLogCacheEndpoints::isTransient)
+                                .doBeforeRetry(
+                                        signal ->
+                                                LOGGER.warn(
+                                                        "Transient error polling Log Cache, retry"
+                                                                + " {}/{}",
+                                                        signal.totalRetries() + 1,
+                                                        MAX_RETRIES,
+                                                        signal.failure()))
+                                .onRetryExhaustedThrow((spec, signal) -> signal.failure()))
+                .map(response -> toPage(response, cursor, request.getDelay()));
+    }
+
+    private static Page toPage(ReadResponse response, long cursor, Duration delay) {
+        List<Envelope> raw =
+                response.getEnvelopes() != null
+                        ? response.getEnvelopes().getBatch()
+                        : Collections.emptyList();
+        long newestAllowed = toNanos(Instant.now().minus(delay));
+
+        List<Envelope> ready = new ArrayList<>();
+        for (Envelope envelope : raw) {
+            if (envelope.getTimestamp() != null && envelope.getTimestamp() <= newestAllowed) {
+                ready.add(envelope);
+            }
+        }
+        ready.sort(Comparator.comparingLong(Envelope::getTimestamp));
+        return new Page(cursor, ready);
+    }
+
+    private static boolean isTransient(Throwable t) {
+        if (t instanceof AbstractCloudFoundryException) {
+            int status = ((AbstractCloudFoundryException) t).getStatusCode();
+            return status == 429 || status >= 500;
+        }
+        return true;
+    }
+
+    private static long toNanos(Instant instant) {
+        return instant.getEpochSecond() * 1_000_000_000L + instant.getNano();
+    }
+
+    private static final class Page {
+
+        private final long cursor;
+
+        private final List<Envelope> envelopes;
+
+        private Page(long cursor, List<Envelope> envelopes) {
+            this.cursor = cursor;
+            this.envelopes = envelopes;
+        }
+
+        private long nextCursor() {
+            return this.envelopes.get(this.envelopes.size() - 1).getTimestamp() + 1;
+        }
+    }
 }

cloudfoundry-client-reactor/src/main/java/org/cloudfoundry/reactor/logcache/v1/_ReactorLogCacheClient.java

@@ -16,16 +16,19 @@
 
 package org.cloudfoundry.reactor.logcache.v1;
 
+import org.cloudfoundry.logcache.v1.Envelope;
 import org.cloudfoundry.logcache.v1.InfoRequest;
 import org.cloudfoundry.logcache.v1.InfoResponse;
 import org.cloudfoundry.logcache.v1.LogCacheClient;
 import org.cloudfoundry.logcache.v1.MetaRequest;
 import org.cloudfoundry.logcache.v1.MetaResponse;
 import org.cloudfoundry.logcache.v1.ReadRequest;
 import org.cloudfoundry.logcache.v1.ReadResponse;
+import org.cloudfoundry.logcache.v1.TailLogsRequest;
 import org.cloudfoundry.reactor.ConnectionContext;
 import org.cloudfoundry.reactor.TokenProvider;
 import org.immutables.value.Value;
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import java.net.URI;
@@ -53,6 +56,11 @@ public Mono<ReadResponse> read(ReadRequest request) {
         return getReactorLogCacheEndpoints().read(request);
     }
 
+    @Override
+    public Flux<Envelope> logsTail(TailLogsRequest request) {
+        return getReactorLogCacheEndpoints().logsTail(request);
+    }
+
     /**
      * The connection context
      */

cloudfoundry-client-reactor/src/main/java/org/cloudfoundry/reactor/util/ErrorPayloadMappers.java

@@ -74,8 +74,11 @@ public static ErrorPayloadMapper fallback() {
                                 ByteBufFlux body =
                                         ByteBufFlux.fromInbound(connection.inbound().receive());
 
+                                // An error response without a body must still surface as an error
+                                // instead of silently completing empty.
                                 return body.aggregate()
                                         .asString()
+                                        .defaultIfEmpty("")
                                         .doFinally(signalType -> connection.channel().close())
                                         .flatMap(
                                                 payload -> {