feat: port contentEncoding, streamed dataset export and raw-bytes log read from apify-client-js

CHANGELOG.md

@@ -5,6 +5,19 @@ All notable changes to the Apify Java client are documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project
 adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [0.9.0] - 2026-10-19
+
+### Added
+
+- `SetRecordOptions.contentEncoding(String)`, matching the reference client's `setRecord`
+  `contentEncoding` option: upload a pre-compressed record value (e.g. a gzipped file) without the
+  client compressing it again.
+- `DatasetClient.streamItems(DownloadItemsFormat, DatasetDownloadOptions)`, resolving to an
+  `InputStream` over the dataset export instead of fully buffering it, so a large export can be
+  processed in chunks without holding all of it in memory.
+- `LogClient.getAsBytes()` / `getAsBytes(LogOptions)`, resolving to the log's raw bytes instead of
+  decoding them as text, matching the reference clients' `buffer`/`get_as_bytes` option.
+
 ## [0.8.0] - 2026-10-12
 
 ### Added

README.md

@@ -23,7 +23,7 @@ Maven (Maven Central is a default repository, so no extra configuration is neede
 <dependency>
   <groupId>com.apify</groupId>
   <artifactId>apify-client</artifactId>
-  <version>0.8.0</version>
+  <version>0.9.0</version>
 </dependency>
 ```
 
@@ -35,7 +35,7 @@ repositories {
 }
 
 dependencies {
-  implementation 'com.apify:apify-client:0.8.0'
+  implementation 'com.apify:apify-client:0.9.0'
 }
 ```
 
@@ -203,7 +203,7 @@ non-blocking. `HttpRequest` and `HttpResponse<T>` below are the JDK's `java.net.
 | Method | Signature | Contract |
 |---|---|---|
 | `sendAsync` | `CompletableFuture<HttpResponse<byte[]>> sendAsync(HttpRequest request)` | Sends the single, fully-prepared request (auth, `User-Agent` and any other headers are already set by the client) and completes the future with the response with its body fully buffered as `byte[]`. Used for every non-streaming call. |
-| `sendStreamingAsync` | `CompletableFuture<HttpResponse<InputStream>> sendStreamingAsync(HttpRequest request)` | Sends the request and completes the future with the response body as a live `InputStream` for incremental consumption, used by log streaming (`LogClient.stream(...)`). The *caller* (the client) is responsible for closing the returned stream; an implementation must not close it itself before completing the future. |
+| `sendStreamingAsync` | `CompletableFuture<HttpResponse<InputStream>> sendStreamingAsync(HttpRequest request)` | Sends the request and completes the future with the response body as a live `InputStream` for incremental consumption, used by log streaming (`LogClient.stream(...)`) and dataset export streaming (`DatasetClient.streamItems(...)`). The *caller* (the client) is responsible for closing the returned stream; an implementation must not close it itself before completing the future. |
 
 Both methods perform exactly one network round-trip each — no retrying, no request mutation
 (retries, auth, and the `User-Agent` header are handled one layer up, by the client itself), and
@@ -317,7 +317,7 @@ try {
 The public `com.apify.client.Version` class (`import com.apify.client.Version;`) exposes two
 constants:
 
-- `Version.CLIENT_VERSION` — the semantic version of this client (`0.8.0`).
+- `Version.CLIENT_VERSION` — the semantic version of this client (`0.9.0`).
 - `Version.API_SPEC_VERSION` — the version of the [Apify OpenAPI specification](https://docs.apify.com/api/openapi.json)
   (its `info.version` field) that this client's endpoints, parameters and models were last generated
   and checked against (`v2-2026-10-08T073338Z`). It is a snapshot, not a live compatibility

docs/misc.md

@@ -80,6 +80,7 @@ Access a build's or run's log directly, or via `client.run(id).log()` / `client.
 | Method | Description |
 |---|---|
 | `get()` / `get(LogOptions)` | The whole log as text. Completes with `Optional<String>` when addressed by an explicit id (`client.log(id)`); on `run.log()`/`build.log()` (no log id of its own), a 404 is ambiguous and throws `NotFoundError` instead — see [Fetching single resources](../README.md#fetching-single-resources). |
+| `getAsBytes()` / `getAsBytes(LogOptions)` | As `get()`, but completes with `Optional<byte[]>` (the raw bytes, not decoded as text). Same 404 handling as `get()`. |
 | `stream()` / `stream(LogOptions)` | A live `InputStream` over the log (for redirection). Completes with `InputStream`; throws on any error response, including a 404. |
 
 `LogOptions` fields: `raw(Boolean)`, `download(Boolean)`.

docs/storages.md

@@ -44,6 +44,7 @@ any field not covered by a typed getter above is still available via the inherit
 | `iterateItems(DatasetListItemsOptions)` / `iterateItems(DatasetListItemsOptions, Long chunkSize)` | A lazy `Flow.Publisher<JsonNode>` over all items; the options' `limit` caps the total yielded (`null`/unset or non-positive = all), the optional `chunkSize` sets the per-request page size (omitted/`null` = server default). |
 | `iterateItems(DatasetListItemsOptions, Long chunkSize, Class<T>)` | As above, decoded into `T`. Returns `Flow.Publisher<T>`. For typed iteration at the server-default page size, pass a `null` chunk size: `iterateItems(opts, null, T.class)`. |
 | `downloadItems(DownloadItemsFormat, DatasetDownloadOptions)` | Serialized bytes. `DownloadItemsFormat` is one of `JSON`, `JSONL`, `CSV`, `XLSX`, `XML`, `RSS`, `HTML`. Completes with `byte[]`. |
+| `streamItems(DownloadItemsFormat, DatasetDownloadOptions)` | As above, but resolves to a live `InputStream` over the export instead of fully buffering it, so a large export is processed in chunks. The caller closes the stream (or reads it to the end). Throws on any error response, including a 404. |
 | `pushItems(Object)` | Push a single item or a list of items. Completes with no value (`CompletableFuture<Void>`). |
 | `getStatistics()` | Dataset statistics. Completes with `JsonNode`; throws `NotFoundError` if the dataset is gone (no separate "statistics absent" state). |
 | `createItemsPublicUrl(DatasetListItemsOptions, Long expiresInSecs)` | A public (optionally signed) items URL, served as `json`. Completes with `String`. |
@@ -66,6 +67,15 @@ PaginationList<JsonNode> page = client.dataset(ds.getId()).listItems(new Dataset
 byte[] csv = client.dataset(ds.getId()).downloadItems(DownloadItemsFormat.CSV, new DatasetDownloadOptions().bom(true)).join();
 ```
 
+For a large export, `streamItems` avoids holding the whole serialized export in memory:
+
+```java
+try (InputStream stream =
+    client.dataset("my-dataset").streamItems(DownloadItemsFormat.CSV, new DatasetDownloadOptions()).join()) {
+  stream.transferTo(System.out);
+}
+```
+
 `DatasetListItemsOptions` fields: `offset` (`Long`, number of items to skip), `limit` (`Long`,
 maximum number of items to return), `desc` (`Boolean`, return items newest-first), `fields`
 (`List<String>`, restrict the output to these source fields), `outputFields` (`List<String>`,
@@ -78,7 +88,8 @@ becomes a separate item), `flatten` (`List<String>`, flatten these nested fields
 keys), `view` (`String`, select a predefined dataset view for field selection), `simplified`
 (`Boolean`, return simplified — flattened and cleaned — items), `skipFailedPages` (`Boolean`, skip
 items that come from failed pages), `signature` (`String`, a pre-shared URL signature granting
-access without an API token). `downloadItems(...)` returns `byte[]` (the serialized export).
+access without an API token). `downloadItems(...)` returns `byte[]` (the serialized export);
+`streamItems(...)` takes the same two arguments but returns a live `InputStream` over it instead.
 `DatasetDownloadOptions` wraps a `DatasetListItemsOptions` (`items(DatasetListItemsOptions)`) and
 adds `attachment` (`Boolean`), `bom` (`Boolean`), `delimiter` (`String`), `skipHeaderRow`
 (`Boolean`), `xmlRoot` (`String`), `xmlRow` (`String`), `feedTitle` (`String`), `feedDescription`
@@ -116,7 +127,7 @@ datasets.
 | `recordExists(String key)` | Whether a record exists. Completes with `boolean`. |
 | `getRecord(String key)` / `getRecord(String key, GetRecordOptions)` | Fetch a record. Completes with `Optional<KeyValueStoreRecord>`. |
 | `setRecord(String key, byte[] value, String contentType)` | Store raw bytes. Completes with no value (`CompletableFuture<Void>`). |
-| `setRecord(String key, byte[] value, String contentType, SetRecordOptions)` | Store raw bytes with write options (`SetRecordOptions`: `timeoutSecs` (`Long`), `doNotRetryTimeouts` (`Boolean`)). Completes with no value. |
+| `setRecord(String key, byte[] value, String contentType, SetRecordOptions)` | Store raw bytes with write options (`SetRecordOptions`: `timeoutSecs` (`Long`), `doNotRetryTimeouts` (`Boolean`), `contentEncoding` (`String`, see below)). Completes with no value. |
 | `setRecordJson(String key, Object value)` | Store JSON. Completes with no value. |
 | `deleteRecord(String key)` | Delete a record. Completes with no value. |
 | `getRecordPublicUrl(String key)` | A public (optionally signed) record URL. Completes with `String`. |
@@ -138,6 +149,27 @@ rec.ifPresent(r -> System.out.println(new String(r.getValue())));
 `GetRecordOptions` (for `getRecord(String key, GetRecordOptions)`) fields: `attachment(Boolean)`
 (request a download disposition) and `signature(String)` (a pre-computed access signature).
 
+`SetRecordOptions.contentEncoding(String)` names an encoding already applied to `value` (e.g.
+`"gzip"` for a pre-compressed file), sent as the `Content-Encoding` header (trimmed first); the
+client then sends the bytes as they are and does not compress them again, and the API accepts
+`gzip`, `br`, `deflate`, or `identity` (matched case-insensitively — `identity` turns the client's
+own automatic compression off for this upload). It must not be blank after trimming. The record is
+stored exactly as uploaded, but since this client sends no `Accept-Encoding`, a later
+`getRecord(...)` transparently receives the decompressed value rather than the stored compressed
+bytes.
+
+```java
+ByteArrayOutputStream buffer = new ByteArrayOutputStream();
+try (java.util.zip.GZIPOutputStream gzip = new java.util.zip.GZIPOutputStream(buffer)) {
+  gzip.write("{\"answer\":42}".getBytes());
+}
+byte[] gzippedReport = buffer.toByteArray();
+
+client.keyValueStore("my-store")
+    .setRecord("report.json", gzippedReport, "application/json", new SetRecordOptions().contentEncoding("gzip"))
+    .join();
+```
+
 `KeyValueStoreRecord` exposes `getKey()`, `getValue()` (raw bytes) and `getContentType()`.
 `KeyValueStoreKeysPage` exposes `getItems()` (a list of `KeyValueStoreKey` with `getKey()`/`getSize()`),
 `isTruncated()`, `getExclusiveStartKey()` and `getNextExclusiveStartKey()`.

pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>com.apify</groupId>
     <artifactId>apify-client</artifactId>
-    <version>0.8.0</version>
+    <version>0.9.0</version>
     <packaging>jar</packaging>
 
     <name>Apify Java Client</name>

src/main/java/com/apify/client/Version.java

@@ -13,7 +13,7 @@ public final class Version {
    * The semantic version of this client library (see <a href="https://semver.org/">SemVer</a>).
    * Changes to the public interface other than additive ones are considered breaking changes.
    */
-  public static final String CLIENT_VERSION = "0.8.0";
+  public static final String CLIENT_VERSION = "0.9.0";
 
   /**
    * The version of the Apify OpenAPI specification this client was generated and verified against.

src/main/java/com/apify/client/dataset/DatasetClient.java

@@ -10,6 +10,7 @@
 import com.apify.client.internal.QueryParams;
 import com.apify.client.internal.ResourceContext;
 import com.apify.client.internal.Signatures;
+import java.io.InputStream;
 import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
@@ -202,11 +203,32 @@ private <T> CompletableFuture<PaginationList<T>> fetchItemsPage(
    */
   public CompletableFuture<byte[]> downloadItems(
       DownloadItemsFormat format, DatasetDownloadOptions options) {
+    String url = itemsUrl(format, options);
+    return http.call("GET", url, null, "", http.baseRequestTimeout()).thenApply(ApiResponse::body);
+  }
+
+  /**
+   * As {@link #downloadItems}, but resolves to the response body as a live {@link InputStream}
+   * instead of fully buffering it, so a large export can be processed in chunks without holding all
+   * of it in memory. The caller is responsible for closing the returned stream (or reading it to
+   * the end). Like {@link com.apify.client.log.LogClient#stream}, this bypasses the
+   * buffered/retrying transport, so a non-2xx response is read to completion and thrown as an
+   * {@link com.apify.client.http.ApifyApiException} rather than retried.
+   */
+  public CompletableFuture<InputStream> streamItems(
+      DownloadItemsFormat format, DatasetDownloadOptions options) {
+    return http.streamOrThrow(itemsUrl(format, options));
+  }
+
+  /**
+   * Builds the export URL shared by {@link #downloadItems} and {@link #streamItems}: the {@code
+   * items} sub-path with {@code format} and the given options applied as query params.
+   */
+  private String itemsUrl(DownloadItemsFormat format, DatasetDownloadOptions options) {
     QueryParams params = new QueryParams();
     params.addString("format", format.wireValue());
     options.apply(params);
-    String url = ctx.mergedParams(params).applyToUrl(ctx.subUrl("items"));
-    return http.call("GET", url, null, "", http.baseRequestTimeout()).thenApply(ApiResponse::body);
+    return ctx.mergedParams(params).applyToUrl(ctx.subUrl("items"));
   }
 
   /**

src/main/java/com/apify/client/internal/HttpClientCore.java

@@ -64,9 +64,6 @@ public final class HttpClientCore {
    */
   private static final int MIN_COMPRESS_BYTES = 1024;
 
-  /** Header announcing the request-body content coding to the server. */
-  private static final String CONTENT_ENCODING_HEADER = "Content-Encoding";
-
   /** {@code Content-Encoding} value for brotli-compressed bodies (matches the reference client). */
   private static final String ENCODING_BROTLI = "br";
 
@@ -190,7 +187,7 @@ public CompletableFuture<ApiResponse> call(
       Compressed compressed = compress(requestBody, BROTLI_AVAILABLE);
       requestBody = compressed.body();
       headers = new LinkedHashMap<>(extraHeaders == null ? Map.of() : extraHeaders);
-      headers.put(CONTENT_ENCODING_HEADER, compressed.encoding());
+      headers.put(HttpHeaders.CONTENT_ENCODING, compressed.encoding());
     }
 
     String path = extractPath(url);
@@ -372,7 +369,7 @@ private static boolean shouldCompress(
     }
     if (extraHeaders != null) {
       for (String key : extraHeaders.keySet()) {
-        if (CONTENT_ENCODING_HEADER.equalsIgnoreCase(key)) {
+        if (HttpHeaders.CONTENT_ENCODING.equalsIgnoreCase(key)) {
           return false;
         }
       }
@@ -672,4 +669,37 @@ public CompletableFuture<HttpResponse<InputStream>> streamAsync(String url) {
             });
     return result;
   }
+
+  /**
+   * As {@link #streamAsync(String)}, additionally inspecting the response itself: a non-2xx status
+   * is drained and thrown as the matching {@link ApifyApiException} subclass (single attempt - this
+   * bypasses the retrying transport, so the attempt number is always 1), while a success response
+   * completes with its live body stream. Shared by every streaming resource method (log streaming,
+   * dataset export streaming) so the drain-and-throw logic lives in one place.
+   */
+  public CompletableFuture<InputStream> streamOrThrow(String url) {
+    return streamAsync(url)
+        .thenApply(
+            resp -> {
+              if (resp.statusCode() >= MAX_SUCCESS_STATUS) {
+                byte[] body = drainQuietly(resp.body());
+                throw buildApiError(resp.statusCode(), body, 1, "GET", extractPath(url));
+              }
+              return resp.body();
+            });
+  }
+
+  /**
+   * Reads a stream to the end and closes it, for capturing a non-2xx streaming response's error
+   * body; a read failure is swallowed (the stream is being drained only to describe an error that
+   * already happened, so a second failure here is not worth surfacing over the original one) and
+   * yields an empty body instead.
+   */
+  private static byte[] drainQuietly(InputStream in) {
+    try (InputStream stream = in) {
+      return stream.readAllBytes();
+    } catch (IOException e) {
+      return new byte[0];
+    }
+  }
 }

src/main/java/com/apify/client/internal/HttpHeaders.java

@@ -16,5 +16,12 @@ public final class HttpHeaders {
   /** The request/response body media-type header name. */
   public static final String CONTENT_TYPE = "Content-Type";
 
+  /**
+   * The request/response body compression header name. Set explicitly on a key-value-store {@code
+   * setRecord} upload, it tells {@link HttpClientCore} the body is already encoded, which both
+   * skips the client's own automatic compression and forwards the caller's encoding verbatim.
+   */
+  public static final String CONTENT_ENCODING = "Content-Encoding";
+
   private HttpHeaders() {}
 }

src/main/java/com/apify/client/internal/ResourceContext.java

@@ -483,8 +483,25 @@ public CompletableFuture<Void> putRaw(
       String contentType,
       Duration timeout,
       boolean doNotRetryTimeouts) {
+    return putRaw(subPath, params, body, contentType, null, timeout, doNotRetryTimeouts);
+  }
+
+  /**
+   * As {@link #putRaw(String, QueryParams, byte[], String, Duration, boolean)}, additionally
+   * setting {@code extraHeaders} on the request (used to forward a caller-supplied {@code
+   * Content-Encoding} on a key-value-store record upload - see {@link HttpHeaders#CONTENT_ENCODING}
+   * - which also stops {@link HttpClientCore} from compressing the already-encoded body again).
+   */
+  public CompletableFuture<Void> putRaw(
+      String subPath,
+      QueryParams params,
+      byte[] body,
+      String contentType,
+      Map<String, String> extraHeaders,
+      Duration timeout,
+      boolean doNotRetryTimeouts) {
     String u = mergedParams(params).applyToUrl(subUrl(subPath));
-    return http.call("PUT", u, body, contentType, timeout, doNotRetryTimeouts)
+    return http.call("PUT", u, body, contentType, extraHeaders, timeout, doNotRetryTimeouts)
         .thenApply(resp -> null);
   }
 

src/main/java/com/apify/client/keyvalue/KeyValueStoreClient.java

@@ -10,6 +10,7 @@
 import com.apify.client.internal.Signatures;
 import java.time.Duration;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Flow;
@@ -308,10 +309,23 @@ public CompletableFuture<Void> setRecord(String key, byte[] value, String conten
 
   /**
    * Stores a record with raw bytes and the given content type, honoring the given write options
-   * ({@code timeoutSecs}, {@code doNotRetryTimeouts}).
+   * ({@code timeoutSecs}, {@code doNotRetryTimeouts}, {@code contentEncoding}).
+   *
+   * @throws IllegalArgumentException if {@code options.contentEncoding()} is set but blank
    */
   public CompletableFuture<Void> setRecord(
       String key, byte[] value, String contentType, SetRecordOptions options) {
+    // Trimmed before validating/forwarding, matching the reference client's
+    // z.string().trim().min(1) schema (it sends the trimmed value, not the original).
+    String contentEncoding = options.contentEncodingValue();
+    if (contentEncoding != null) {
+      contentEncoding = contentEncoding.trim();
+      if (contentEncoding.isEmpty()) {
+        throw new IllegalArgumentException("contentEncoding must not be blank");
+      }
+    }
+    Map<String, String> extraHeaders =
+        contentEncoding == null ? null : Map.of(HttpHeaders.CONTENT_ENCODING, contentEncoding);
     Duration timeout =
         options.timeoutSecsValue() != null
             ? Duration.ofSeconds(options.timeoutSecsValue())
@@ -321,6 +335,7 @@ public CompletableFuture<Void> setRecord(
         new QueryParams(),
         value,
         contentType,
+        extraHeaders,
         timeout,
         options.doNotRetryTimeoutsValue());
   }