diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 45edad25..13ebdd6f 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -36,6 +36,13 @@ jobs:
         python-version: ${{ matrix.python-version }}
         cache: pip
 
+    # Apache Tika 4 requires Java 17+; test against the minimum
+    - name: Set up Java 17
+      uses: actions/setup-java@v6
+      with:
+        distribution: temurin
+        java-version: "17"
+
     - name: Install dependencies
       run: |
         python -m pip install --upgrade pip
diff --git a/README.md b/README.md
index 3a5582ec..04e75b85 100755
--- a/README.md
+++ b/README.md
@@ -11,14 +11,21 @@ tika-python
 
 A Python port of the [Apache Tika](http://tika.apache.org/)
 library that makes Tika available using the
-[Tika REST Server](https://cwiki.apache.org/confluence/display/TIKA/TikaServer).
+[Tika REST Server](https://tika.apache.org/docs/4.1.x/using-tika/server/index.html).
 
 This makes Apache Tika available as a Python library,
 installable via Setuptools, Pip and Easy Install.
 
-To use this library, you need to have Java 11+ installed on your
+To use this library, you need to have Java 17+ installed on your
 system as tika-python starts up the Tika REST server in the
-background.
+background. Apache Tika 4 runs its parsers in forked worker JVMs that
+are started with the `java` found on the `PATH` (when `TIKA_JAVA`
+points at a specific binary, its directory is put on the server's
+`PATH` for you).
+
+tika-python 4.x targets Apache Tika 4.x and does not work against a
+Tika 3 server; stay on tika-python 3.x for that. See
+"Upgrading from Tika 3.x" below for what changed.
 
 Inspired by [Aptivate Tika](https://github.com/aptivate/python-tika).
 
@@ -28,30 +35,83 @@ Installation (with pip)
 
 Airgap Environment Setup
 ------------------------
-To get this working in a disconnected environment, download a tika server file (both tika-server.jar and tika-server.jar.md5, which can be found [here](https://repo1.maven.org/maven2/org/apache/tika/tika-server-standard/)) and set the TIKA_SERVER_JAR environment variable to TIKA_SERVER_JAR="file:///<yourpath>/tika-server-standard.jar" which successfully tells `python-tika` to "download" this file and move it to `/tmp/tika-server-standard.jar` and run as background process.
+To get this working in a disconnected environment, download the Tika server
+distribution `tika-server-standard-<version>.zip` (and, optionally, its
+`.sha512` checksum file) from the
+[Apache archive](https://archive.apache.org/dist/tika/) and point
+`TIKA_SERVER_JAR` at it:
+
+```bash
+export TIKA_SERVER_JAR="file:///<yourpath>/tika-server-standard-4.1.0.zip"
+# or simply
+export TIKA_SERVER_JAR="/<yourpath>/tika-server-standard-4.1.0.zip"
+```
+
+`python-tika` verifies the zip against the `.sha512` file if it sits next to it
+(otherwise it logs a warning and skips the check), unpacks it once into
+`TIKA_PATH` and starts the server from there. You can also unzip the
+distribution yourself and point `TIKA_SERVER_JAR` at the
+`tika-server-standard-<version>.jar` inside the unzipped directory; that jar
+only runs next to its `lib/` directory.
 
-This is the only way to run `python-tika` without internet access. Without this set, the default is to check the tika version and pull latest every time from Apache.
+This is the only way to run `python-tika` without internet access. Without this
+set, the default is to download the release matching `TIKA_VERSION` from
+`dlcdn.apache.org`, falling back to `archive.apache.org` for releases that are
+no longer on the mirrors.
 
 Environment Variables
 ---------------------
 These are read once, when tika/tika.py is initially loaded and used throughout after that.
 
-1. `TIKA_VERSION` - set to the version string, e.g., 1.12 or default to current Tika version.
-2. `TIKA_SERVER_JAR` - set to the full URL to the remote Tika server jar to download and cache.
-3. `TIKA_SERVER_ENDPOINT` - set to the host (local or remote) for the running Tika server jar.
+1. `TIKA_VERSION` - set to the version string, e.g., 4.1.0 or default to current Tika version.
+2. `TIKA_SERVER_JAR` - URL, `file://` URL or local path of the `tika-server-standard-<version>.zip` distribution to download and cache. May also point at a `tika-server-standard-<version>.jar` that you already unzipped next to its `lib/` directory.
+3. `TIKA_SERVER_ENDPOINT` - set to the host (local or remote) for the running Tika server.
 4. `TIKA_CLIENT_ONLY` - if set to True, then `TIKA_SERVER_JAR` is ignored, and relies on the value for `TIKA_SERVER_ENDPOINT` and treats Tika like a REST client.
-3. `TIKA_JAR_HASH_ALGO` - set to `sha1` when running on FIPS-compliant systems; default value is `md5`.
-4. `TIKA_SERVER_ENDPOINT` - set to the host (local or remote) for the running Tika server jar.
-5. `TIKA_CLIENT_ONLY` - if set to True, then `TIKA_SERVER_JAR` is ignored, and relies on the value for `TIKA_SERVER_ENDPOINT` and treats Tika like a REST client.
-6. `TIKA_TRANSLATOR` - set to the fully qualified class name (defaults to Lingo24) for the Tika translator implementation.
-7. `TIKA_SERVER_CLASSPATH` - set to a string (delimited by ':' for each additional path) to prepend to the Tika server jar path.
-8. `TIKA_LOG_PATH` - set to a directory with write permissions and the `tika.log` and `tika-server.log` files will be placed in this directory.
-9. `TIKA_PATH` - set to a directory with write permissions and the `tika_server.jar` file will be placed in this directory.
-10. `TIKA_JAVA` - set the Java runtime name, e.g., `java` or `java9`
-11. `TIKA_STARTUP_SLEEP` - number of seconds (`float`) to wait per check if Tika server is launched at runtime
-12. `TIKA_STARTUP_MAX_RETRY` - number of checks (`int`) to attempt for Tika server startup if launched at runtime
-13. `TIKA_JAVA_ARGS` - set java runtime arguments, e.g, `-Xmx4g`
-14. `TIKA_LOG_FILE` - set the filename for the log file. default: `tika.log`. if it is an empty string (`''`), no log file is created.
+5. `TIKA_JAR_HASH_ALGO` - hash used to verify the downloaded distribution; default `sha512`, which is the only checksum Apache publishes for the zip (and is FIPS compliant).
+6. `TIKA_SERVER_CLASSPATH` - set to a string (delimited by ':' for each additional path) to prepend to the Tika server jar path.
+7. `TIKA_LOG_PATH` - set to a directory with write permissions and the `tika.log` and `tika-server.log` files will be placed in this directory.
+8. `TIKA_PATH` - set to a directory with write permissions; the server zip, its checksum and the unzipped `tika-server-standard-<version>/` directory are placed there (default: the system temp directory).
+9. `TIKA_JAVA` - set the Java runtime name or path, e.g., `java` or `/usr/lib/jvm/java-17/bin/java`; Java 17 or later is required.
+10. `TIKA_STARTUP_SLEEP` - number of seconds (`float`) to wait per check if Tika server is launched at runtime (default 5)
+11. `TIKA_STARTUP_MAX_RETRY` - number of checks (`int`) to attempt for Tika server startup if launched at runtime (default 6)
+12. `TIKA_JAVA_ARGS` - set java runtime arguments for the server JVM, e.g, `-Xmx4g`. Since Tika 4 parsing happens in forked worker JVMs whose options (`forkedJvmArgs`, `javaPath`, `numClients`) are set in the server's JSON config file, not through this variable.
+13. `TIKA_LOG_FILE` - set the filename for the log file. default: `tika.log`. if it is an empty string (`''`), no log file is created.
+
+Upgrading from Tika 3.x
+=======================
+tika-python 4.x moved from Apache Tika 3.3 to Apache Tika 4.1. Tika 4 changed
+its server in ways that are visible to users of this library:
+
+* **Java 17+** is required (Tika 3 ran on Java 11).
+* **Metadata keys were renamed** and the server does not map them back:
+  `X-TIKA:content` is now `tk:content`, `resourceName` is `tk:resource-name`,
+  `X-TIKA:Parsed-By` is `tk:parsed-by`, `X-TIKA:EXCEPTION:*` is `tk:exception:*`,
+  and many format specific keys are kebab-case now (e.g. `pdf:pdf-version`).
+  `Content-Type` and `Content-Length` are unchanged. `parsed["content"]` keeps
+  working; code reading other metadata keys needs updating.
+* **The server distribution is a zip**, not a runnable jar. `TIKA_SERVER_JAR`
+  now points at `tika-server-standard-<version>.zip`, which is verified with its
+  `.sha512` checksum and unzipped into `TIKA_PATH`. Leftover `tika-server.jar*`
+  files from 3.x in your temp directory can be deleted.
+* **The translate interface was removed** (`tika.translate` and the `translate`
+  CLI command); Tika 4 has no `/translate` endpoint.
+* **Language detection** works on extracted text: `language.from_file` parses
+  the document first. Tika 4 also uses a different detector, which is less
+  reliable on very short snippets.
+* **Unpack returns embedded documents** under `attachments`, keyed by the
+  server's entry names (`1.<ext>`, `2.<ext>`, ...) rather than the original
+  file names.
+* `parser.from_file(..., service='text')` uses `/tika/text` (plain `/tika`
+  returns Markdown in Tika 4), and `service='text', xmlContent=True` now really
+  returns XHTML.
+* **`config_path` must be a JSON file** containing a `server` element; Tika 4
+  no longer reads `tika-config.xml`. Per-request `X-Tika-*` headers are
+  silently ignored by Tika 4.
+* New response codes: 429 when the server's worker pool is saturated, 503 when a
+  worker crashed or timed out, 413 for requests over 1 GiB. Error bodies are
+  JSON; `parsed["status"]` carries the code and `metadata`/`content` are `None`.
+* Memory limits for parsing are set in the server's JSON config
+  (`forkedJvmArgs`), no longer through `TIKA_JAVA_ARGS`.
 
 Testing it out
 ==============
@@ -146,14 +206,22 @@ print(parsed["content"])
 Unpack Interface
 ----------------
 The unpack interface handles both metadata and text extraction in a single
-call and internally returns back a tarball of metadata and text entries that
-is internally unpacked, reducing the wire load for extraction.
+call and additionally returns the documents embedded in the container (images
+in a PDF, members of an archive, attachments of an email, ...). The server
+answers with a zip that is unpacked internally.
 
 ```python
 from tika import unpack
 parsed = unpack.from_file('/path/to/file')
+print(parsed["metadata"])     # metadata of the container document
+print(parsed["content"])      # its extracted text
+for name, data in parsed["attachments"].items():
+    print(name, len(data))    # embedded documents, e.g. '1.png' -> bytes
 ```
 
+Attachment names are the entry names chosen by the server (`1.<ext>`, `2.<ext>`, ...);
+the container document itself is not part of `attachments`.
+
 Detect Interface
 ----------------------
 The detect interface provides a IANA MIME type classification for the
@@ -180,21 +248,14 @@ print(config.getDetectors())
 Language Detection Interface
 ---------------------------------
 The language detection interface provides a 2 character language
-code texted based on the text in provided file.
+code based on the text in the provided file. The server's `/language`
+endpoint expects text, so `from_file` first extracts the text of the
+document and then detects its language.
 
 ```python
 from tika import language
 print(language.from_file('/path/to/file'))
-```
-
-Translate Interface
-------------------------
-The translate interface translates the text automatically extracted
-by Tika from the source language to the destination language.
-
-```python
-from tika import translate
-print(translate.from_file('/path/to/spanish', 'es', 'en'))
+print(language.from_buffer('A few sentences of text are enough to tell the language apart.'))
 ```
 
 Using a Buffer
@@ -253,7 +314,7 @@ print parsed["metadata"]
 
 Customizing the Tika Server Request
 ---------------------------
-You may customize the outgoing HTTP request to Tika server by setting `requestOptions` on the `.from_file` and `.from_buffer` methods (Parser, Unpack , Detect, Config, Language, Translate). It should be a dictionary of arguments that will be passed to the request method. The [request method documentation](https://requests.kennethreitz.org/en/master/api/#requests.request) specifies valid arguments. This will override any defaults except for `url` and `params `/`data`.
+You may customize the outgoing HTTP request to Tika server by setting `requestOptions` on the `.from_file` and `.from_buffer` methods (Parser, Unpack, Detect, Config, Language). It should be a dictionary of arguments that will be passed to the request method. The [request method documentation](https://requests.kennethreitz.org/en/master/api/#requests.request) specifies valid arguments. This will override any defaults except for `url` and `params `/`data`.
 
 ```python
 from tika import parser
@@ -268,18 +329,17 @@ directory.
 
 The options and help for the command line tool can be seen by typing
 `tika-python` without any arguments. This will also download a copy of
-the tika-server jar and start it if you haven't done so already.
+the Tika server distribution and start it if you haven't done so already.
 
 ```bash
-tika.py [-v] [-o <outputDir>] [--server <TikaServerEndpoint>] [--install <UrlToTikaServerJar>] [--port <portNumber>] <command> <option> <urlOrPathToFile>
+tika.py [-v] [-e] [-o <outputDir>] [--server <TikaServerEndpoint>] [--install <UrlToTikaServerZip>] [--port <portNumber>] <command> <option> <urlOrPathToFile>
 
 tika.py parse all test.pdf test2.pdf                   (write output JSON metadata files for test1.pdf_meta.json and test2.pdf_meta.json)
 tika.py detect type test.pdf                           (returns mime-type as text/plain)
 tika.py language file french.txt                       (returns language e.g., fr as text/plain)
-tika.py translate fr:en french.txt                     (translates the file french.txt from french to english)
 tika.py config mime-types                              (see what mime-types the Tika Server can handle)
 
-A simple python and command-line client for Tika using the standalone Tika server (JAR file).
+A simple python and command-line client for Tika using the standalone Tika server (zip distribution).
 All commands return results in JSON format by default (except text in text/plain).
 
 To parse docs, use:
@@ -292,7 +352,6 @@ Commands:
   parse  = parse the input file and write a JSON doc file.ext_meta.json containing the extracted metadata, text, or both
   detect type = parse the stream and 'detect' the MIME/media type, return in text/plain
   language file = parse the file stream and identify the language of the text, return its 2 character code in text/plain
-  translate src:dest = parse and extract text and then translate the text from source language to destination language
   config = return a JSON doc describing the configuration of the Tika server (i.e. mime-types it
              can handle, or installed detectors or parsers)
 
@@ -304,7 +363,7 @@ Switches:
   --encode, -e           = encode response in UTF-8
   --csv, -c    = report detect output in comma-delimited format
   --server <TikaServerEndpoint>  = use a remote Tika Server at this endpoint, otherwise use local server
-  --install <UrlToTikaServerJar> = download and exec Tika Server (JAR file), starting server on default port 9998
+  --install <UrlToTikaServerZip> = download, unpack and exec the Tika Server distribution, starting server on default port 9998
 
 Example usage as python client:
 -- from tika import runCommand, parse1
diff --git a/tests/test_language.py b/tests/test_language.py
index 703aee5f..63f724aa 100644
--- a/tests/test_language.py
+++ b/tests/test_language.py
@@ -13,4 +13,9 @@ def test_local_path(test_file_path):
 
 
 def test_local_buffer():
-    assert language.from_buffer("Good evening, David. How are you?") == "en"
+    assert (
+        language.from_buffer(
+            "Good evening, David. How are you? The weather has been lovely this week and the garden is finally in bloom."
+        )
+        == "en"
+    )
diff --git a/tests/test_parser.py b/tests/test_parser.py
index 72d9db81..f1a8cc9c 100644
--- a/tests/test_parser.py
+++ b/tests/test_parser.py
@@ -1,10 +1,60 @@
 # SPDX-License-Identifier: Apache-2.0
 
+import json
 from http import HTTPStatus
 
 from tika import parser
 
 
+# --- response parsing, no server needed -------------------------------------------
+
+
+def test_parse_concatenates_content_and_merges_metadata():
+    body = json.dumps(
+        [
+            {parser._CONTENT_KEY: "outer ", "Content-Type": "application/pdf", "dc:title": "t"},
+            {parser._CONTENT_KEY: "inner", "Content-Type": "image/png"},
+        ]
+    )
+    parsed = parser._parse((200, body))
+    assert parsed["status"] == 200
+    assert parsed["content"] == "outer inner"
+    assert parsed["metadata"]["Content-Type"] == ["application/pdf", "image/png"]
+    assert parsed["metadata"]["dc:title"] == "t"
+    assert parser._CONTENT_KEY not in parsed["metadata"]
+
+
+def test_parse_without_content_yields_none():
+    parsed = parser._parse((200, json.dumps([{"Content-Type": "image/png"}])))
+    assert parsed["content"] is None
+    assert parsed["metadata"] == {"Content-Type": "image/png"}
+
+
+def test_parse_meta_service():
+    parsed = parser._parse((200, json.dumps({"Content-Type": "application/pdf"})), service="meta")
+    assert parsed == {"metadata": {"Content-Type": "application/pdf"}, "content": None, "status": 200}
+
+
+def test_parse_text_service_returns_body_verbatim():
+    parsed = parser._parse((200, "plain text"), service="text")
+    assert parsed == {"metadata": None, "content": "plain text", "status": 200}
+
+
+def test_parse_empty_body():
+    assert parser._parse((204, "")) == {"metadata": None, "content": None, "status": 204}
+    assert parser._parse(None) == {"metadata": None, "content": None}
+
+
+def test_parse_does_not_decode_error_bodies():
+    parsed = parser._parse((503, json.dumps({"status": "TIMEOUT", "message": "worker timed out"})))
+    assert parsed == {"metadata": None, "content": None, "status": 503}
+    parsed = parser._parse((422, "Unprocessable Entity"), service="meta")
+    assert parsed == {"metadata": None, "content": None, "status": 422}
+
+
+# --- integration, against the live server -----------------------------------------
+
+
 def test_remote_pdf(remote_fixture_base_url):
     """parse remote PDF"""
     assert parser.from_file(f"{remote_fixture_base_url}/remote.pdf")
diff --git a/tests/test_server_install.py b/tests/test_server_install.py
new file mode 100644
index 00000000..f4b7d42c
--- /dev/null
+++ b/tests/test_server_install.py
@@ -0,0 +1,377 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Unit tests for locating, downloading, verifying and extracting the Tika server.
+
+These run without a Tika server, Java or network access.
+"""
+
+import hashlib
+import os
+import subprocess
+import zipfile
+from pathlib import Path
+from unittest import mock
+
+import pytest
+import requests
+
+import tika.tika
+from tika.tika import TikaDownloadError, TikaException
+
+DIST_URL = "https://dlcdn.apache.org/tika/9.9.9/tika-server-standard-9.9.9.zip"
+ARCHIVE_URL = "https://archive.apache.org/dist/tika/9.9.9/tika-server-standard-9.9.9.zip"
+SHA512 = hashlib.sha512(b"x").hexdigest()
+
+
+def make_dist_zip(path, version="9.9.9", top_level_dir=None):
+    prefix = f"{top_level_dir}/" if top_level_dir else ""
+    Path(path).parent.mkdir(parents=True, exist_ok=True)
+    with zipfile.ZipFile(path, "w") as archive:
+        archive.writestr(f"{prefix}tika-server-standard-{version}.jar", b"launcher")
+        archive.writestr(f"{prefix}lib/tika-core-{version}.jar", b"core")
+        archive.writestr(f"{prefix}plugins/tika-pipes-file-system/plugin.jar", b"plugin")
+        archive.writestr(f"{prefix}LICENSE", b"Apache-2.0")
+    return path
+
+
+def sha512_of(path):
+    return hashlib.sha512(Path(path).read_bytes()).hexdigest()
+
+
+@pytest.fixture
+def install_env(tmp_path, monkeypatch):
+    """Point the module-level cache config at a scratch directory."""
+    monkeypatch.setattr(tika.tika, "TikaJarPath", str(tmp_path / "cache"))
+    monkeypatch.setattr(tika.tika, "TikaJarHashAlgo", "sha512")
+    return tmp_path
+
+
+class FakeDownloader:
+    """Stand-in for ``_urlretrieve`` serving an in-memory set of files by URL."""
+
+    def __init__(self, files=None, missing=()):
+        self.files = dict(files or {})
+        self.missing = set(missing)
+        self.requested = []
+
+    def __call__(self, url, filename, **kwargs):
+        self.requested.append(url)
+        if url in self.missing or url not in self.files:
+            raise TikaDownloadError(f"Failed to download {url}: 404", 404)
+        Path(filename).parent.mkdir(parents=True, exist_ok=True)
+        Path(filename).write_bytes(self.files[url])
+        return filename
+
+
+# --- pure helpers -----------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    "text",
+    [
+        SHA512,
+        SHA512.upper(),
+        SHA512 + "\n",
+        f"{SHA512}  tika-server-standard-9.9.9.zip\n",
+        "tika-server-standard-9.9.9.zip: "
+        + " ".join(SHA512[i : i + 4].upper() for i in range(0, 64, 4))
+        + "\n                                "
+        + " ".join(SHA512[i : i + 4].upper() for i in range(64, 128, 4))
+        + "\n",
+    ],
+)
+def test_parse_checksum_file_accepts_common_layouts(text):
+    assert tika.tika._parseChecksumFile(text, "sha512") == SHA512
+
+
+@pytest.mark.parametrize("text", ["", "not a digest", SHA512[:-2], f"{SHA512}ab"])
+def test_parse_checksum_file_rejects_garbage(text):
+    with pytest.raises(TikaException):
+        tika.tika._parseChecksumFile(text, "sha512")
+
+
+def test_parse_checksum_file_honours_algorithm():
+    md5 = hashlib.md5(b"x").hexdigest()
+    assert tika.tika._parseChecksumFile(md5, "md5") == md5
+    with pytest.raises(TikaException):
+        tika.tika._parseChecksumFile(md5, "sha512")
+
+
+@pytest.mark.parametrize(
+    "url, expected",
+    [
+        (DIST_URL, ARCHIVE_URL),
+        ("https://downloads.apache.org/tika/9.9.9/tika-server-standard-9.9.9.zip", ARCHIVE_URL),
+        (ARCHIVE_URL, None),
+        ("https://repo1.maven.org/maven2/org/apache/tika/x.jar", None),
+        ("https://dlcdn.apache.org/solr/9.0.0/solr.tgz", None),
+        ("/tmp/tika-server-standard-9.9.9.zip", None),
+    ],
+)
+def test_fallback_distribution_url(url, expected):
+    assert tika.tika._fallbackDistributionUrl(url) == expected
+
+
+def test_local_path_from_url_or_path(tmp_path):
+    target = tmp_path / "tika-server-standard-9.9.9.zip"
+    assert tika.tika._localPathFromUrlOrPath(str(target)) == str(target)
+    assert tika.tika._localPathFromUrlOrPath(target.as_uri()) == str(target)
+    relative = tika.tika._localPathFromUrlOrPath("dist/x.zip")
+    assert os.path.isabs(relative) and relative.endswith("x.zip")
+    windows = tika.tika._localPathFromUrlOrPath(r"C:\tika\x.zip")
+    assert windows is not None and windows.endswith("x.zip")
+    assert tika.tika._localPathFromUrlOrPath(DIST_URL) is None
+    assert tika.tika._localPathFromUrlOrPath("http://localhost:9998/x.zip") is None
+
+
+@pytest.mark.parametrize(
+    "output, expected",
+    [
+        ('openjdk version "17.0.2" 2022-01-18\nOpenJDK Runtime Environment', 17),
+        ('java version "1.8.0_292"\nJava(TM) SE Runtime Environment', 8),
+        ('openjdk version "25.0.4.1" 2026-08-18', 25),
+        ('Picked up JAVA_TOOL_OPTIONS: -Dhttp.proxyHost=x\nopenjdk version "21" 2023-09-19', 21),
+        ("gibberish", None),
+    ],
+)
+def test_java_major_version(output, expected):
+    completed = subprocess.CompletedProcess(["java", "-version"], 0, stdout="", stderr=output)
+    with mock.patch("tika.tika.subprocess.run", return_value=completed):
+        assert tika.tika._javaMajorVersion("java") == expected
+
+
+def test_java_major_version_missing_binary():
+    with mock.patch("tika.tika.subprocess.run", side_effect=FileNotFoundError):
+        assert tika.tika._javaMajorVersion("/nonexistent/java") is None
+
+
+def test_jar_needs_lib_dir(tmp_path):
+    thin = tmp_path / "thin.jar"
+    with zipfile.ZipFile(thin, "w") as jar:
+        jar.writestr(
+            "META-INF/MANIFEST.MF",
+            "Manifest-Version: 1.0\r\nMain-Class: org.apache.tika.server.core.TikaServerCli\r\n"
+            "Class-Path: lib/tika-server-core-9.9.9.jar lib/tika-core-9.9.9.jar lib/very-long-nam\r\n"
+            " e-that-wraps.jar\r\n",
+        )
+    fat = tmp_path / "fat.jar"
+    with zipfile.ZipFile(fat, "w") as jar:
+        jar.writestr("META-INF/MANIFEST.MF", "Manifest-Version: 1.0\r\nMain-Class: x.Main\r\n")
+    assert tika.tika._jarNeedsLibDir(str(thin)) is True
+    assert tika.tika._jarNeedsLibDir(str(fat)) is False
+    assert tika.tika._jarNeedsLibDir(str(tmp_path / "missing.jar")) is False
+
+
+# --- extraction --------------------------------------------------------------------
+
+
+def test_extract_server_distribution(tmp_path):
+    archive = make_dist_zip(tmp_path / "dist.zip")
+    install_dir = tmp_path / "cache" / "tika-server-standard-9.9.9"
+
+    tika.tika._extractServerDistribution(str(archive), str(install_dir))
+
+    assert (install_dir / "tika-server-standard-9.9.9.jar").read_bytes() == b"launcher"
+    assert (install_dir / "lib" / "tika-core-9.9.9.jar").is_file()
+    assert (install_dir / "plugins" / "tika-pipes-file-system" / "plugin.jar").is_file()
+    assert os.listdir(install_dir.parent) == ["tika-server-standard-9.9.9"], "no temp dirs left behind"
+    assert tika.tika._findServerJar(str(install_dir)) == str(install_dir / "tika-server-standard-9.9.9.jar")
+
+
+def test_extract_server_distribution_unwraps_single_top_level_dir(tmp_path):
+    archive = make_dist_zip(tmp_path / "dist.zip", top_level_dir="tika-server-standard-9.9.9")
+    install_dir = tmp_path / "tika-server-standard-9.9.9"
+    tika.tika._extractServerDistribution(str(archive), str(install_dir))
+    assert tika.tika._findServerJar(str(install_dir)) == str(install_dir / "tika-server-standard-9.9.9.jar")
+
+
+def test_extract_server_distribution_keeps_existing_install(tmp_path):
+    archive = make_dist_zip(tmp_path / "dist.zip")
+    install_dir = tmp_path / "tika-server-standard-9.9.9"
+    tika.tika._extractServerDistribution(str(archive), str(install_dir))
+    (install_dir / "marker").write_text("first")
+    tika.tika._extractServerDistribution(str(archive), str(install_dir))
+    assert (install_dir / "marker").read_text() == "first"
+    assert sorted(os.listdir(tmp_path)) == ["dist.zip", "tika-server-standard-9.9.9"], "no temp dirs left behind"
+
+
+def test_extract_server_distribution_bad_zip_leaves_nothing(tmp_path):
+    archive = tmp_path / "dist.zip"
+    archive.write_bytes(b"not a zip")
+    install_dir = tmp_path / "tika-server-standard-9.9.9"
+    with pytest.raises(zipfile.BadZipFile):
+        tika.tika._extractServerDistribution(str(archive), str(install_dir))
+    assert not install_dir.exists()
+    assert os.listdir(tmp_path) == ["dist.zip"]
+
+
+def test_find_server_jar_requires_lib_dir(tmp_path):
+    (tmp_path / "tika-server-standard-9.9.9.jar").write_bytes(b"x")
+    assert tika.tika._findServerJar(str(tmp_path)) is None
+    (tmp_path / "lib").mkdir()
+    assert tika.tika._findServerJar(str(tmp_path)) == str(tmp_path / "tika-server-standard-9.9.9.jar")
+    assert tika.tika._findServerJar(str(tmp_path / "missing")) is None
+
+
+def test_find_server_jar_accepts_single_unnamed_jar(tmp_path):
+    (tmp_path / "lib").mkdir()
+    (tmp_path / "custom-build.jar").write_bytes(b"x")
+    assert tika.tika._findServerJar(str(tmp_path)) == str(tmp_path / "custom-build.jar")
+    (tmp_path / "another.jar").write_bytes(b"x")
+    assert tika.tika._findServerJar(str(tmp_path)) is None
+
+
+# --- download layer ----------------------------------------------------------------
+
+
+def test_urlretrieve_reports_http_status(tmp_path):
+    response = mock.Mock()
+    response.raise_for_status.side_effect = requests.HTTPError(response=mock.Mock(status_code=404))
+    target = tmp_path / "x.zip"
+    with mock.patch("tika.tika.requests.get", return_value=response):
+        with pytest.raises(TikaDownloadError) as excinfo:
+            tika.tika._urlretrieve("https://example.org/x.zip", str(target))
+    assert excinfo.value.status_code == 404
+    assert not target.exists()
+
+
+def test_urlretrieve_connection_error_has_no_status(tmp_path):
+    with mock.patch("tika.tika.requests.get", side_effect=requests.ConnectionError("refused")):
+        with pytest.raises(TikaDownloadError) as excinfo:
+            tika.tika._urlretrieve("https://example.org/x.zip", str(tmp_path / "x.zip"))
+    assert excinfo.value.status_code is None
+
+
+def test_download_with_fallback_retries_on_archive(tmp_path, monkeypatch):
+    downloader = FakeDownloader({ARCHIVE_URL: b"zip"}, missing={DIST_URL})
+    monkeypatch.setattr(tika.tika, "_urlretrieve", downloader)
+    target = tmp_path / "x.zip"
+    tika.tika._downloadWithFallback(DIST_URL, str(target))
+    assert downloader.requested == [DIST_URL, ARCHIVE_URL]
+    assert target.read_bytes() == b"zip"
+
+
+def test_download_with_fallback_does_not_retry_other_urls(tmp_path, monkeypatch):
+    url = "https://repo1.maven.org/maven2/org/apache/tika/x.jar"
+    downloader = FakeDownloader(missing={url})
+    monkeypatch.setattr(tika.tika, "_urlretrieve", downloader)
+    with pytest.raises(TikaDownloadError):
+        tika.tika._downloadWithFallback(url, str(tmp_path / "x.jar"))
+    assert downloader.requested == [url]
+
+
+# --- ensureTikaServerJar -----------------------------------------------------------
+
+
+@pytest.fixture
+def remote_dist(install_env, tmp_path, monkeypatch):
+    """A fake remote hosting a valid zip and its sha512 sidecar under DIST_URL."""
+    archive = make_dist_zip(tmp_path / "upstream.zip")
+    downloader = FakeDownloader(
+        {DIST_URL: archive.read_bytes(), DIST_URL + ".sha512": (sha512_of(archive) + "\n").encode()}
+    )
+    monkeypatch.setattr(tika.tika, "_urlretrieve", downloader)
+    return downloader
+
+
+def test_ensure_downloads_verifies_and_extracts(install_env, remote_dist):
+    cache = install_env / "cache"
+
+    jar = tika.tika.ensureTikaServerJar(DIST_URL)
+
+    assert jar == str(cache / "tika-server-standard-9.9.9" / "tika-server-standard-9.9.9.jar")
+    assert (cache / "tika-server-standard-9.9.9" / "lib").is_dir()
+    assert (cache / "tika-server-standard-9.9.9.zip").is_file(), "archive kept for later re-extraction"
+    assert (cache / "tika-server-standard-9.9.9.zip.sha512").is_file()
+    assert remote_dist.requested == [DIST_URL, DIST_URL + ".sha512"]
+
+
+def test_ensure_reuses_extracted_distribution(install_env, remote_dist):
+    first = tika.tika.ensureTikaServerJar(DIST_URL)
+    remote_dist.requested.clear()
+    with mock.patch("tika.tika._fileDigest", side_effect=AssertionError("must not hash again")):
+        assert tika.tika.ensureTikaServerJar(DIST_URL) == first
+    assert remote_dist.requested == []
+
+
+def test_ensure_redownloads_once_on_checksum_mismatch(install_env, remote_dist):
+    remote_dist.files[DIST_URL + ".sha512"] = (SHA512 + "\n").encode()  # wrong digest
+    with pytest.raises(TikaException, match="Checksum mismatch"):
+        tika.tika.ensureTikaServerJar(DIST_URL)
+    assert remote_dist.requested == [DIST_URL, DIST_URL + ".sha512", DIST_URL, DIST_URL + ".sha512"]
+    assert not (install_env / "cache" / "tika-server-standard-9.9.9").exists()
+
+
+def test_ensure_falls_back_to_archive(install_env, remote_dist):
+    remote_dist.files[ARCHIVE_URL] = remote_dist.files.pop(DIST_URL)
+    remote_dist.files[ARCHIVE_URL + ".sha512"] = remote_dist.files.pop(DIST_URL + ".sha512")
+
+    jar = tika.tika.ensureTikaServerJar(DIST_URL)
+
+    assert jar.endswith("tika-server-standard-9.9.9.jar")
+    assert remote_dist.requested == [DIST_URL, ARCHIVE_URL, DIST_URL + ".sha512", ARCHIVE_URL + ".sha512"]
+
+
+def test_ensure_local_zip_via_file_url(install_env, tmp_path, monkeypatch):
+    archive = make_dist_zip(tmp_path / "offline" / "tika-server-standard-9.9.9.zip")
+    (tmp_path / "offline" / "tika-server-standard-9.9.9.zip.sha512").write_text(sha512_of(archive))
+    monkeypatch.setattr(tika.tika, "_urlretrieve", mock.Mock(side_effect=AssertionError("no network")))
+
+    jar = tika.tika.ensureTikaServerJar(archive.as_uri())
+
+    assert jar == str(install_env / "cache" / "tika-server-standard-9.9.9" / "tika-server-standard-9.9.9.jar")
+    assert archive.is_file(), "the user's archive is left in place"
+    assert not (install_env / "cache" / "tika-server-standard-9.9.9.zip").exists(), "not copied"
+
+
+def test_ensure_local_zip_plain_path_without_sidecar_warns(install_env, tmp_path, monkeypatch, caplog):
+    archive = make_dist_zip(tmp_path / "offline" / "tika-server-standard-9.9.9.zip")
+    monkeypatch.setattr(tika.tika, "_urlretrieve", mock.Mock(side_effect=AssertionError("no network")))
+
+    with caplog.at_level("WARNING", logger="tika.tika"):
+        jar = tika.tika.ensureTikaServerJar(str(archive))
+
+    assert jar.endswith("tika-server-standard-9.9.9.jar")
+    assert "skipping verification" in caplog.text
+
+
+def test_ensure_local_zip_with_bad_sidecar_fails_without_deleting(install_env, tmp_path):
+    archive = make_dist_zip(tmp_path / "offline" / "tika-server-standard-9.9.9.zip")
+    (tmp_path / "offline" / "tika-server-standard-9.9.9.zip.sha512").write_text(SHA512)
+
+    with pytest.raises(TikaException, match="Checksum mismatch"):
+        tika.tika.ensureTikaServerJar(str(archive))
+
+    assert archive.is_file()
+    assert not (install_env / "cache" / "tika-server-standard-9.9.9").exists()
+
+
+def test_ensure_local_jar_override_is_used_as_is(install_env, tmp_path):
+    unpacked = tmp_path / "unpacked"
+    (unpacked / "lib").mkdir(parents=True)
+    jar = unpacked / "tika-server-standard-9.9.9.jar"
+    jar.write_bytes(b"launcher")
+
+    assert tika.tika.ensureTikaServerJar(str(jar)) == str(jar)
+    assert tika.tika.ensureTikaServerJar(jar.as_uri()) == str(jar)
+    with pytest.raises(TikaException, match="does not exist"):
+        tika.tika.ensureTikaServerJar(str(tmp_path / "missing.jar"))
+
+
+def test_ensure_remote_jar_override_is_downloaded_and_verified(install_env, monkeypatch):
+    url = "https://repo1.maven.org/maven2/org/apache/tika/tika-server-standard/3.3.2/tika-server-standard-3.3.2.jar"
+    downloader = FakeDownloader({url: b"fat jar", url + ".sha512": hashlib.sha512(b"fat jar").hexdigest().encode()})
+    monkeypatch.setattr(tika.tika, "_urlretrieve", downloader)
+
+    jar = tika.tika.ensureTikaServerJar(url)
+
+    assert jar == str(install_env / "cache" / "tika-server-standard-3.3.2.jar")
+    assert Path(jar).read_bytes() == b"fat jar"
+    assert downloader.requested == [url, url + ".sha512"]
+    downloader.requested.clear()
+    assert tika.tika.ensureTikaServerJar(url) == jar
+    assert downloader.requested == [], "cached jar is reused"
+
+
+def test_ensure_rejects_url_without_file_name(install_env):
+    with pytest.raises(TikaException, match="file name"):
+        tika.tika.ensureTikaServerJar("https://dlcdn.apache.org/tika/")
diff --git a/tests/test_unpack.py b/tests/test_unpack.py
index 04a6fd43..06279594 100644
--- a/tests/test_unpack.py
+++ b/tests/test_unpack.py
@@ -1,5 +1,9 @@
 # SPDX-License-Identifier: Apache-2.0
 
+import io
+import json
+import zipfile
+
 from tika import unpack
 
 # Test data
@@ -7,12 +11,59 @@
 TEXT_ASCII = "Hello, world!!"
 
 
+# --- zip response parsing, no server needed ---------------------------------------
+
+
+def make_unpack_zip(entries):
+    buffer = io.BytesIO()
+    with zipfile.ZipFile(buffer, "w") as archive:
+        for name, data in entries.items():
+            archive.writestr(name, json.dumps(data) if isinstance(data, dict) else data)
+    return buffer.getvalue()
+
+
+def test_parse_zip_splits_container_and_attachments():
+    body = make_unpack_zip(
+        {
+            "0.pdf": b"%PDF-original",
+            "0.pdf.metadata.json": {"tk:content": "hello", "Content-Type": "application/pdf"},
+            "1.png": b"\x89PNG",
+            "1.png.metadata.json": {"Content-Type": "image/png"},
+            "2.docx": b"PK",
+        }
+    )
+    parsed = unpack._parse((200, body))
+    assert parsed["content"] == "hello"
+    assert parsed["metadata"] == {"Content-Type": "application/pdf"}
+    assert parsed["attachments"] == {"1.png": b"\x89PNG", "2.docx": b"PK"}
+
+
+def test_parse_zip_accepts_stem_named_sidecar():
+    body = make_unpack_zip({"0.txt": b"hi", "0.metadata.json": {"tk:content": "hi", "Content-Type": "text/plain"}})
+    parsed = unpack._parse((200, body))
+    assert parsed["content"] == "hi"
+    assert parsed["metadata"]["Content-Type"] == "text/plain"
+    assert parsed["attachments"] == {}
+
+
+def test_parse_zip_without_sidecar():
+    parsed = unpack._parse((200, make_unpack_zip({"0.txt": b"hi"})))
+    assert parsed == {"content": "", "metadata": {}, "attachments": {}}
+
+
+def test_parse_zip_empty_response():
+    assert unpack._parse((204, b"")) == {}
+    assert unpack._parse((200, None)) == {}
+    assert unpack._parse(None) == {}
+
+
 def test_utf8(tmp_path):
     """Test UTF-8 encoding"""
     test_file = tmp_path / "test_utf8.txt"
     test_file.write_bytes(TEXT_UTF8.encode("utf8"))
     parsed = unpack.from_file(str(test_file))
     assert parsed["content"].strip() == TEXT_UTF8
+    assert parsed["attachments"] == {}
 
 
 def test_ascii(tmp_path):
@@ -21,6 +72,21 @@ def test_ascii(tmp_path):
     test_file.write_text(TEXT_ASCII)
     parsed = unpack.from_file(str(test_file))
     assert parsed["content"].strip() == TEXT_ASCII
+    assert parsed["metadata"]["Content-Type"].startswith("text/plain")
+    assert parsed["attachments"] == {}
+
+
+def test_embedded_documents_become_attachments(tmp_path):
+    """A zip is a container: its members come back as attachments, its own text as content"""
+    container = tmp_path / "container.zip"
+    with zipfile.ZipFile(container, "w") as archive:
+        archive.writestr("a.txt", "alpha text")
+        archive.writestr("b.txt", "bravo text")
+    parsed = unpack.from_file(str(container))
+    assert parsed["metadata"]["Content-Type"] == "application/zip"
+    assert "a.txt" in parsed["content"] and "b.txt" in parsed["content"]
+    assert sorted(parsed["attachments"].values()) == [b"alpha text", b"bravo text"]
+    assert all(name.startswith(("1.", "2.")) for name in parsed["attachments"]), parsed["attachments"].keys()
 
 
 def test_from_buffer():
@@ -28,6 +94,8 @@ def test_from_buffer():
     assert parsed is not None
     assert parsed["metadata"] is not None
     assert parsed["metadata"]["Content-Length"] == "5"
+    assert parsed["content"].strip() == "what?"
+    assert parsed["attachments"] == {}
 
 
 def test_from_buffer_with_headers():
diff --git a/tika/__init__.py b/tika/__init__.py
index 7e0a03b5..079856e0 100644
--- a/tika/__init__.py
+++ b/tika/__init__.py
@@ -13,7 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-__version__ = "3.3.3.dev0"
+__version__ = "4.1.0.dev0"
 
 from pkgutil import extend_path
 
diff --git a/tika/detector.py b/tika/detector.py
index b561545d..66a22b22 100644
--- a/tika/detector.py
+++ b/tika/detector.py
@@ -32,6 +32,6 @@ def from_buffer(string, config_path=None, requestOptions={}):
     :param string: buffered content whose type needs to be detected
     :return:
     '''
-    status, response = callServer('put', ServerEndpoint, '/detect/stream', string,
+    status, response = callServer('put', ServerEndpoint, '/detect', string,
                                   {'Accept': 'text/plain'}, False, config_path=config_path, requestOptions=requestOptions)
     return response
diff --git a/tika/language.py b/tika/language.py
index 7f38c960..039d8d39 100644
--- a/tika/language.py
+++ b/tika/language.py
@@ -32,6 +32,6 @@ def from_buffer(string, requestOptions={}):
     :param string: buffered data
     :return:
     '''
-    status, response = callServer('put', ServerEndpoint, '/language/string', string,
+    status, response = callServer('put', ServerEndpoint, '/language', string,
                                   {'Accept': 'text/plain'}, False, requestOptions=requestOptions)
     return response
diff --git a/tika/parser.py b/tika/parser.py
index 118477bd..e0abf9f3 100644
--- a/tika/parser.py
+++ b/tika/parser.py
@@ -18,6 +18,10 @@
 
 from .tika import ServerEndpoint, callServer, parse1
 
+# metadata key under which the /rmeta endpoint returns the extracted text of each document
+# (X-TIKA:content before Tika 4.0)
+_CONTENT_KEY = 'tk:content'
+
 
 def from_file(filename, serverEndpoint=ServerEndpoint, service='all', xmlContent=False, headers=None, config_path=None, requestOptions={}, raw_response=False):
     '''
@@ -38,7 +42,7 @@ def from_file(filename, serverEndpoint=ServerEndpoint, service='all', xmlContent
     if not xmlContent:
         output = parse1(service, filename, serverEndpoint, headers=headers, config_path=config_path, requestOptions=requestOptions)
     else:
-        output = parse1(service, filename, serverEndpoint, services={'meta': '/meta', 'text': '/tika', 'all': '/rmeta/xml'},
+        output = parse1(service, filename, serverEndpoint, services={'meta': '/meta', 'text': '/tika/xml', 'all': '/rmeta/xml'},
                             headers=headers, config_path=config_path, requestOptions=requestOptions)
     if raw_response:
         return output
@@ -92,6 +96,10 @@ def _parse(output, service='all'):
         parsed["content"] = output[1]
         return parsed
 
+    if output[0] != 200:
+        # an error body (JSON since Tika 4, plain text before) is not a parse result
+        return parsed
+
     realJson = json.loads(output[1])
 
     parsed["metadata"] = {}
@@ -102,8 +110,8 @@ def _parse(output, service='all'):
 
     content = ""
     for js in realJson:
-        if "X-TIKA:content" in js:
-            content += js["X-TIKA:content"]
+        if _CONTENT_KEY in js:
+            content += js[_CONTENT_KEY]
 
     if content == "":
         content = None
@@ -112,7 +120,7 @@ def _parse(output, service='all'):
 
     for js in realJson:
         for n in js:
-            if n != "X-TIKA:content":
+            if n != _CONTENT_KEY:
                 if n in parsed["metadata"]:
                     if not isinstance(parsed["metadata"][n], list):
                         parsed["metadata"][n] = [parsed["metadata"][n]]
diff --git a/tika/tika.py b/tika/tika.py
index fa351f10..e084f2fa 100755
--- a/tika/tika.py
+++ b/tika/tika.py
@@ -38,35 +38,27 @@
     from tika import language
     print(language.from_file('/path/to/file'))
 
-**Use Tika Translate**::
+**Tika-Python Configuration**
 
-   from tika import translate
-   print(translate.from_file('/path/to/file', 'srcLang', 'destLang')
-   # Use auto Language detection feature
-   print(translate.from_file('/path/to/file', 'destLang')
+You can use a custom server configuration file: a JSON file with a ``server`` element since
+Apache Tika 4 (see https://tika.apache.org/docs/4.1.x/ for the format). Configuration is set
+the first time the server is started. To use a configuration file with a parser or detector::
 
-***Tika-Python Configuration***
-You can now use custom configuration files. See https://tika.apache.org/1.18/configuring.html
-for details on writing configuration files. Configuration is set the first time the server is started.
-To use a configuration file with a parser, or detector:
     parsed = parser.from_file('/path/to/file', config_path='/path/to/configfile')
-or:
     detected = detector.from_file('/path/to/file', config_path='/path/to/configfile')
-or:
     detected = detector.from_buffer('some buffered content', config_path='/path/to/configfile')
 
 '''
 
 USAGE = """
-tika.py [-v] [-e] [-o <outputDir>] [--server <TikaServerEndpoint>] [--install <UrlToTikaServerJar>] [--port <portNumber>] <command> <option> <urlOrPathToFile>
+tika.py [-v] [-e] [-o <outputDir>] [--server <TikaServerEndpoint>] [--install <UrlToTikaServerZip>] [--port <portNumber>] <command> <option> <urlOrPathToFile>
 
 tika.py parse all test.pdf test2.pdf                   (write output JSON metadata files for test1.pdf_meta.json and test2.pdf_meta.json)
 tika.py detect type test.pdf                           (returns mime-type as text/plain)
 tika.py language file french.txt                       (returns language e.g., fr as text/plain)
-tika.py translate fr:en french.txt                     (translates the file french.txt from french to english)
 tika.py config mime-types                              (see what mime-types the Tika Server can handle)
 
-A simple python and command-line client for Tika using the standalone Tika server (JAR file).
+A simple python and command-line client for Tika using the standalone Tika server (zip distribution).
 All commands return results in JSON format by default (except text in text/plain).
 
 To parse docs, use:
@@ -79,7 +71,6 @@
   parse  = parse the input file and write a JSON doc file.ext_meta.json containing the extracted metadata, text, or both
   detect type = parse the stream and 'detect' the MIME/media type, return in text/plain
   language file = parse the file stream and identify the language of the text, return its 2 character code in text/plain
-  translate src:dest = parse and extract text and then translate the text from source language to destination language
   config = return a JSON doc describing the configuration of the Tika server (i.e. mime-types it
              can handle, or installed detectors or parsers)
 
@@ -91,7 +82,7 @@
   --encode, -e           = encode response in UTF-8
   --csv, -c    = report detect output in comma-delimited format
   --server <TikaServerEndpoint>  = use a remote Tika Server at this endpoint, otherwise use local server
-  --install <UrlToTikaServerJar> = download and exec Tika Server (JAR file), starting server on default port 9998
+  --install <UrlToTikaServerZip> = download, unpack and exec the Tika Server distribution, starting server on default port 9998
 
 Example usage as python client:
 -- from tika import runCommand, parse1
@@ -123,11 +114,15 @@ def make_content_disposition_header(fn):
 import io
 import logging
 import platform
+import shutil
 import signal
 import socket
+import subprocess
 import tempfile
+import zipfile
 from os import walk
 from subprocess import STDOUT, Popen
+from urllib.request import url2pathname
 
 import requests
 
@@ -152,25 +147,38 @@ def make_content_disposition_header(fn):
 log.setLevel(logging.INFO)
 
 Windows = True if platform.system() == "Windows" else False
-TikaVersion = os.getenv('TIKA_VERSION', '3.3.2')
+TikaVersion = os.getenv('TIKA_VERSION', '4.1.0')
 TikaJarPath = os.getenv('TIKA_PATH', tempfile.gettempdir())
 TikaFilesPath = tempfile.gettempdir()
 TikaServerLogFilePath = log_path
+# Apache Tika 4 ships the runnable server as a zip distribution in the Apache dist area;
+# current releases live on the CDN, every release stays on the archive.
+TikaDistBaseUrl = 'https://dlcdn.apache.org/tika/'
+TikaArchiveBaseUrl = 'https://archive.apache.org/dist/tika/'
 TikaServerJar = os.getenv(
     'TIKA_SERVER_JAR',
-    "http://search.maven.org/remotecontent?filepath=org/apache/tika/tika-server-standard/"+TikaVersion+"/tika-server-standard-"+TikaVersion+".jar")
-TikaJarHashAlgo=os.getenv('TIKA_JAR_HASH_ALGO', 'md5')
+    TikaDistBaseUrl + TikaVersion + '/tika-server-standard-' + TikaVersion + '.zip')
+TikaJarHashAlgo = os.getenv('TIKA_JAR_HASH_ALGO', 'sha512')
+
+
+def _tikaMajorVersion(version):
+    try:
+        return int(str(version).split('.')[0])
+    except ValueError:
+        return None
+
+
+# Tika 4.x needs Java 17; older servers (still usable via TIKA_VERSION/TIKA_SERVER_JAR) ran on 11
+TikaMinJavaVersion = 11 if (_tikaMajorVersion(TikaVersion) or 4) < 4 else 17
 ServerHost = "localhost"
 Port = "9998"
 ServerEndpoint = os.getenv(
     'TIKA_SERVER_ENDPOINT', 'http://' + ServerHost + ':' + Port)
-Translator = os.getenv(
-    'TIKA_TRANSLATOR',
-    "org.apache.tika.language.translate.Lingo24Translator")
 TikaClientOnly = os.getenv('TIKA_CLIENT_ONLY', False)
 TikaServerClasspath = os.getenv('TIKA_SERVER_CLASSPATH', '')
 TikaStartupSleep = float(os.getenv('TIKA_STARTUP_SLEEP', 5))
-TikaStartupMaxRetry = int(os.getenv('TIKA_STARTUP_MAX_RETRY', 3))
+# Tika 4 boots its parse worker pool before reporting "Started"; allow up to ~30 s on slow machines
+TikaStartupMaxRetry = int(os.getenv('TIKA_STARTUP_MAX_RETRY', 6))
 TikaJava = os.getenv("TIKA_JAVA", "java")
 TikaJavaArgs = os.getenv("TIKA_JAVA_ARGS", '')
 
@@ -184,6 +192,25 @@ def make_content_disposition_header(fn):
 class TikaException(Exception):
     pass
 
+# Accept header expected by each /tika output path (Tika 4 no longer picks the format from Accept)
+_TikaAcceptTypes = {
+    '/tika': 'text/plain',
+    '/tika/text': 'text/plain',
+    '/tika/md': 'text/plain',
+    '/tika/xml': 'text/xml',
+    '/tika/html': 'text/html',
+    '/tika/json': 'application/json',
+}
+
+class TikaDownloadError(RuntimeError):
+    '''
+    Raised when a download fails. ``status_code`` holds the HTTP status when the
+    server answered, ``None`` for connection level failures.
+    '''
+    def __init__(self, message, status_code=None):
+        super().__init__(message)
+        self.status_code = status_code
+
 def echo2(*s): sys.stderr.write(str('tika.py: %s\n') % str(' ').join(map(str, s)))
 def warn(*s):  echo2('Warn:', *s)
 def die(*s):   warn('Error:',  *s); echo2(USAGE); sys.exit()
@@ -193,7 +220,7 @@ def runCommand(cmd, option, urlOrPaths, port, outDir=None,
                verbose=Verbose, encode=EncodeUtf8):
     '''
     Run the Tika command by calling the Tika server and return results in JSON format (or plain text).
-    :param cmd: a command from set ``{'parse', 'detect', 'language', 'translate', 'config'}``
+    :param cmd: a command from set ``{'parse', 'detect', 'language', 'config'}``
     :param option:
     :param urlOrPaths:
     :param port:
@@ -216,7 +243,8 @@ def runCommand(cmd, option, urlOrPaths, port, outDir=None,
     elif cmd == "language":
         return detectLang(option, urlOrPaths, serverEndpoint, verbose, tikaServerJar)
     elif cmd == "translate":
-        return doTranslate(option, urlOrPaths, serverEndpoint, verbose, tikaServerJar)
+        log.error('The translate command was removed: Apache Tika 4 no longer offers /translate endpoints.')
+        raise TikaException('translate was removed with Apache Tika 4; use tika-python 3.x against a Tika 3 server')
     elif cmd == "config":
         status, resp = getConfig(option, serverEndpoint, verbose, tikaServerJar)
         return resp
@@ -247,7 +275,7 @@ def getPaths(urlOrPaths):
 
 def parseAndSave(option, urlOrPaths, outDir=None, serverEndpoint=ServerEndpoint, verbose=Verbose, tikaServerJar=TikaServerJar,
                  responseMimeType='application/json', metaExtension='_meta.json',
-                 services={'meta': '/meta', 'text': '/tika', 'all': '/rmeta'}):
+                 services={'meta': '/meta', 'text': '/tika/text', 'all': '/rmeta/xml'}):
     '''
     Parse the objects and write extracted metadata and/or text in JSON format to matching
     filename with an extension of '_meta.json'.
@@ -279,7 +307,7 @@ def parseAndSave(option, urlOrPaths, outDir=None, serverEndpoint=ServerEndpoint,
 
 def parse(option, urlOrPaths, serverEndpoint=ServerEndpoint, verbose=Verbose, tikaServerJar=TikaServerJar,
           responseMimeType='application/json',
-          services={'meta': '/meta', 'text': '/tika', 'all': '/rmeta'}, rawResponse=False):
+          services={'meta': '/meta', 'text': '/tika/text', 'all': '/rmeta/xml'}, rawResponse=False):
     '''
     Parse the objects and return extracted metadata and/or text in JSON format.
     :param option:
@@ -296,7 +324,7 @@ def parse(option, urlOrPaths, serverEndpoint=ServerEndpoint, verbose=Verbose, ti
 
 def parse1(option, urlOrPath, serverEndpoint=ServerEndpoint, verbose=Verbose, tikaServerJar=TikaServerJar,
           responseMimeType='application/json',
-          services={'meta': '/meta', 'text': '/tika', 'all': '/rmeta/text'}, rawResponse=False, headers=None, config_path=None, requestOptions={}):
+          services={'meta': '/meta', 'text': '/tika/text', 'all': '/rmeta/text'}, rawResponse=False, headers=None, config_path=None, requestOptions={}):
     '''
     Parse the object and return extracted metadata and/or text in JSON format.
     :param option:
@@ -313,13 +341,13 @@ def parse1(option, urlOrPath, serverEndpoint=ServerEndpoint, verbose=Verbose, ti
     headers = headers or {}
 
     path, file_type = getRemoteFile(urlOrPath, TikaFilesPath)
-    headers.update({'Accept': responseMimeType, 'Content-Disposition': make_content_disposition_header(path.encode('utf-8') if type(path) is str else path)})
 
     if option not in services:
         log.warning('config option must be one of meta, text, or all; using all.')
     service = services.get(option, services['all'])
-    if service == '/tika': responseMimeType = 'text/plain'
-    headers.update({'Accept': responseMimeType, 'Content-Disposition': make_content_disposition_header(path.encode('utf-8') if type(path) is str else path)})
+    # Tika 4 selects the output format by path, not by Accept, and rejects a mismatching Accept with 406
+    responseMimeType = _TikaAcceptTypes.get(service, responseMimeType)
+    headers.update({'Accept': responseMimeType, 'Content-Disposition': make_content_disposition_header(path)})
     with urlOrPath if _is_file_object(urlOrPath) else open(path, 'rb') as f:
         status, response = callServer('put', serverEndpoint, service, f,
                                       headers, verbose, tikaServerJar, config_path=config_path,
@@ -330,7 +358,7 @@ def parse1(option, urlOrPath, serverEndpoint=ServerEndpoint, verbose=Verbose, ti
 
 def detectLang(option, urlOrPaths, serverEndpoint=ServerEndpoint, verbose=Verbose, tikaServerJar=TikaServerJar,
                 responseMimeType='text/plain',
-                services={'file' : '/language/stream'}):
+                services={'file' : '/language'}):
     '''
     Detect the language of the provided stream and return its 2 character code as text/plain.
     :param option:
@@ -348,9 +376,11 @@ def detectLang(option, urlOrPaths, serverEndpoint=ServerEndpoint, verbose=Verbos
 
 def detectLang1(option, urlOrPath, serverEndpoint=ServerEndpoint, verbose=Verbose, tikaServerJar=TikaServerJar,
                responseMimeType='text/plain',
-               services={'file' : '/language/stream'}, requestOptions={}):
+               services={'file' : '/language'}, requestOptions={}):
     '''
-    Detect the language of the provided stream and return its 2 character code as text/plain.
+    Detect the language of the provided document and return its 2 character code as text/plain.
+    Since Tika 4 the /language endpoint expects text rather than a document, so the text is
+    extracted first (via /tika/text) and then submitted for detection.
     :param option:
     :param urlOrPath:
     :param serverEndpoint:
@@ -360,73 +390,20 @@ def detectLang1(option, urlOrPath, serverEndpoint=ServerEndpoint, verbose=Verbos
     :param services:
     :return:
     '''
-    path, mode = getRemoteFile(urlOrPath, TikaFilesPath)
     if option not in services:
         log.exception('Language option must be one of %s ' % bytes(services.keys()))
         raise TikaException('Language option must be one of %s ' % bytes(services.keys()))
     service = services[option]
-    status, response = callServer('put', serverEndpoint, service, open(path, 'rb'),
+    status, text = parse1('text', urlOrPath, serverEndpoint, verbose, tikaServerJar, requestOptions=requestOptions)
+    if status != 200 or not text:
+        return (status, text)
+    status, response = callServer('put', serverEndpoint, service, text,
             {'Accept': responseMimeType}, verbose, tikaServerJar, requestOptions=requestOptions)
     return (status, response)
 
-def doTranslate(option, urlOrPaths, serverEndpoint=ServerEndpoint, verbose=Verbose, tikaServerJar=TikaServerJar,
-                responseMimeType='text/plain',
-                services={'all': '/translate/all'}):
-    '''
-    Translate the file from source language to destination language.
-    :param option:
-    :param urlOrPaths:
-    :param serverEndpoint:
-    :param verbose:
-    :param tikaServerJar:
-    :param responseMimeType:
-    :param services:
-    :return:
-    '''
-    paths = getPaths(urlOrPaths)
-    return [doTranslate1(option, path, serverEndpoint, verbose, tikaServerJar, responseMimeType, services)
-            for path in paths]
-
-def doTranslate1(option, urlOrPath, serverEndpoint=ServerEndpoint, verbose=Verbose, tikaServerJar=TikaServerJar,
-                 responseMimeType='text/plain',
-                 services={'all': '/translate/all'}, requestOptions={}):
-    '''
-
-    :param option:
-    :param urlOrPath:
-    :param serverEndpoint:
-    :param verbose:
-    :param tikaServerJar:
-    :param responseMimeType:
-    :param services:
-    :return:
-    '''
-    path, mode = getRemoteFile(urlOrPath, TikaFilesPath)
-    srcLang = ""
-    destLang = ""
-
-    if ":" in option:
-        options = option.rsplit(':')
-        srcLang = options[0]
-        destLang = options[1]
-        if len(options) != 2:
-            log.exception('Translate options are specified as srcLang:destLang or as destLang')
-            raise TikaException('Translate options are specified as srcLang:destLang or as destLang')
-    else:
-        destLang = option
-
-    if srcLang != "" and destLang != "":
-        service = services["all"] + "/" + Translator + "/" + srcLang + "/" + destLang
-    else:
-        service = services["all"] + "/" + Translator + "/" + destLang
-    status, response = callServer('put', serverEndpoint, service, open(path, 'rb'),
-                                  {'Accept' : responseMimeType},
-                                  verbose, tikaServerJar, requestOptions=requestOptions)
-    return (status, response)
-
 def detectType(option, urlOrPaths, serverEndpoint=ServerEndpoint, verbose=Verbose, tikaServerJar=TikaServerJar,
                responseMimeType='text/plain',
-               services={'type': '/detect/stream'}):
+               services={'type': '/detect'}):
     '''
     Detect the MIME/media type of the stream and return it in text/plain.
     :param option:
@@ -444,7 +421,7 @@ def detectType(option, urlOrPaths, serverEndpoint=ServerEndpoint, verbose=Verbos
 
 def detectType1(option, urlOrPath, serverEndpoint=ServerEndpoint, verbose=Verbose, tikaServerJar=TikaServerJar,
                responseMimeType='text/plain',
-               services={'type': '/detect/stream'}, config_path=None, requestOptions={}):
+               services={'type': '/detect'}, config_path=None, requestOptions={}):
     '''
     Detect the MIME/media type of the stream and return it in text/plain.
     :param option:
@@ -464,7 +441,7 @@ def detectType1(option, urlOrPath, serverEndpoint=ServerEndpoint, verbose=Verbos
     status, response = callServer('put', serverEndpoint, service, open(path, 'rb'),
             {
                 'Accept': responseMimeType,
-                'Content-Disposition': make_content_disposition_header(path.encode('utf-8') if type(path) is str else path)
+                'Content-Disposition': make_content_disposition_header(path)
             },
             verbose, tikaServerJar, config_path=config_path, requestOptions=requestOptions)
     if csvOutput == 1:
@@ -556,7 +533,8 @@ def callServer(verb, serverEndpoint, service, data, headers, verbose=Verbose, ti
 
 def checkTikaServer(scheme="http", serverHost=ServerHost, port=Port, tikaServerJar=TikaServerJar, classpath=None, config_path=None):
     '''
-    Check that tika-server is running.  If not, download JAR file and start it up.
+    Check that tika-server is running.  If not, fetch the server distribution (see
+    ``ensureTikaServerJar``) and start it up.
     :param scheme: e.g. http or https
     :param serverHost:
     :param port:
@@ -569,20 +547,12 @@ def checkTikaServer(scheme="http", serverHost=ServerHost, port=Port, tikaServerJ
     if port is None:
         port = '443' if scheme == 'https' else '80'
 
-    urlp = urlparse(tikaServerJar)
     serverEndpoint = '%s://%s:%s' % (scheme, serverHost, port)
-    jarPath = os.path.join(TikaJarPath, 'tika-server.jar')
     if 'localhost' in serverEndpoint or '127.0.0.1' in serverEndpoint:
         alreadyRunning = checkPortIsOpen(serverHost, port)
 
         if not alreadyRunning:
-            if not os.path.isfile(jarPath) and urlp.scheme != '':
-                getRemoteJar(tikaServerJar, jarPath)
-
-            if not checkJarSig(tikaServerJar, jarPath):
-                os.remove(jarPath)
-                tikaServerJar = getRemoteJar(tikaServerJar, jarPath)
-
+            jarPath = ensureTikaServerJar(tikaServerJar)
             status = startServer(jarPath, TikaJava, TikaJavaArgs, serverHost, port, classpath, config_path)
             if not status:
                 log.error("Failed to receive startup confirmation from startServer.")
@@ -591,22 +561,33 @@ def checkTikaServer(scheme="http", serverHost=ServerHost, port=Port, tikaServerJ
 
 def checkJarSig(tikaServerJar, jarPath):
     '''
-    Checks the signature of Jar
-    :param tikaServerJar:
-    :param jarPath:
-    :return: ``True`` if the signature of the jar matches
+    Verifies a downloaded server archive (or jar) against its ``.<TIKA_JAR_HASH_ALGO>``
+    checksum sidecar. For a remote source the sidecar is downloaded next to ``jarPath``
+    when it is missing. For a local source (plain path or ``file://`` URL) the sidecar is
+    looked up next to the source; when there is none the check is skipped with a warning
+    so that air-gapped installs keep working.
+    :param tikaServerJar: URL or local path the archive came from
+    :param jarPath: local path of the archive to verify
+    :return: ``True`` if the digest matches (or no sidecar is available for a local source)
     '''
     localChecksumPath = '.'.join([jarPath, TikaJarHashAlgo])
     if not os.path.isfile(localChecksumPath):
-        remoteChecksum = '.'.join([tikaServerJar, TikaJarHashAlgo])
-        getRemoteJar(remoteChecksum, localChecksumPath)
-    m = hashlib.new(TikaJarHashAlgo)
-    with open(jarPath, 'rb') as f:
-        binContents = f.read()
-        m.update(binContents)
-        with open(f"{jarPath}.{TikaJarHashAlgo}", "r") as em:
-            existingContents = em.read()
-            return existingContents == m.hexdigest()
+        sourcePath = _localPathFromUrlOrPath(tikaServerJar)
+        if sourcePath is None:
+            remoteChecksum = '.'.join([tikaServerJar, TikaJarHashAlgo])
+            getRemoteJar(remoteChecksum, localChecksumPath)
+        else:
+            localChecksumPath = '.'.join([sourcePath, TikaJarHashAlgo])
+            if not os.path.isfile(localChecksumPath):
+                log.warning('No %s checksum found next to %s; skipping verification.', TikaJarHashAlgo, sourcePath)
+                return True
+    with open(localChecksumPath, 'r') as em:
+        try:
+            expected = _parseChecksumFile(em.read(), TikaJarHashAlgo)
+        except TikaException as e:
+            log.warning('%s: %s', localChecksumPath, e)
+            return False
+    return expected == _fileDigest(jarPath, TikaJarHashAlgo)
 
 
 def startServer(tikaServerJar, java_path = TikaJava, java_args = TikaJavaArgs, serverHost = ServerHost, port = Port, classpath=None, config_path=None):
@@ -651,20 +632,30 @@ def startServer(tikaServerJar, java_path = TikaJava, java_args = TikaJavaArgs, s
         log.error("Unable to create tika-server.log at %s due to permission error." % (TikaServerLogFilePath))
         return False
 
-    # Check that specified java binary is available on path
-    try:
-        _ = Popen(java_path, stdout=open(os.devnull, "w"), stderr=open(os.devnull, "w"))
-    except FileNotFoundError as e:
-        log.error("Unable to run java; is it installed?")
+    # Check that the specified java binary is available and recent enough
+    javaVersion = _javaMajorVersion(java_path)
+    if javaVersion is None:
+        log.error("Unable to run java (%s); is it installed?", java_path)
         return False
+    if javaVersion < TikaMinJavaVersion:
+        log.error("Apache Tika %s needs Java %d or later, found Java %d at %s",
+                  TikaVersion, TikaMinJavaVersion, javaVersion, java_path)
+        return False
+
+    # Tika 4 forks worker JVMs that resolve "java" from PATH; point them at the same runtime
+    env = os.environ.copy()
+    javaDir = os.path.dirname(java_path)
+    if javaDir:
+        env['PATH'] = javaDir + os.pathsep + env.get('PATH', '')
 
     # Run java with jar args
     global TikaServerProcess
+    log.info("Starting Tika server: %s", cmd_string)
     # Patch for Windows support
     if Windows:
-        TikaServerProcess = Popen(cmd_string, stdout=logFile, stderr=STDOUT, shell=True, start_new_session=True)
+        TikaServerProcess = Popen(cmd_string, stdout=logFile, stderr=STDOUT, shell=True, start_new_session=True, env=env)
     else:
-        TikaServerProcess = Popen(cmd_string, stdout=logFile, stderr=STDOUT, shell=True, preexec_fn=os.setsid)
+        TikaServerProcess = Popen(cmd_string, stdout=logFile, stderr=STDOUT, shell=True, preexec_fn=os.setsid, env=env)
 
     # Check logs and retry as configured
     try_count = 0
@@ -682,10 +673,25 @@ def startServer(tikaServerJar, java_path = TikaJava, java_args = TikaJavaArgs, s
 
     if not is_started:
         log.error("Tika startup log message not received after %d tries." % (TikaStartupMaxRetry))
+        _logServerLogTail(tika_log_file_path)
         return False
     else:
         return True
 
+
+def _logServerLogTail(logPath, lines=20):
+    '''
+    Logs the last lines of the server log so a failed startup can be diagnosed without
+    hunting for tika-server.log.
+    '''
+    try:
+        with open(logPath, 'r', errors='replace') as f:
+            tail = f.readlines()[-lines:]
+    except OSError:
+        return
+    if tail:
+        log.error("Last %d lines of %s:\n%s", len(tail), logPath, ''.join(tail).rstrip())
+
 def killServer():
     '''
     Kills the tika server started by the current execution instance
@@ -746,7 +752,7 @@ def _urlretrieve(
         The filepath where the file was saved.
 
     Raises:
-        requests.RequestException: If the download fails.
+        TikaDownloadError: If the download fails (``status_code`` set for HTTP errors).
         IOError: If there's an issue writing to the file.
     """
     headers = {"user-agent": "tika-python"}
@@ -775,7 +781,8 @@ def _urlretrieve(
         # Clean up partial file on error
         if os.path.exists(filename):
             os.remove(filename)
-        raise RuntimeError(f"Failed to download {url}: {e}") from e
+        status_code = getattr(getattr(e, 'response', None), 'status_code', None)
+        raise TikaDownloadError(f"Failed to download {url}: {e}", status_code) from e
 
 
 def getRemoteFile(urlOrPath, destPath):
@@ -803,18 +810,242 @@ def getRemoteFile(urlOrPath, destPath):
 
 def getRemoteJar(urlOrPath, destPath):
     '''
-    Fetches URL to local path or just return absolute path.
-    :param urlOrPath: remote resource locator
+    Fetches URL to local path or just return the local path.
+    :param urlOrPath: remote resource locator, ``file://`` URL or local path
     :param destPath: Path to store the resource, usually a path on file system
     :return: tuple having (path, 'local'/'remote')
     '''
+    localPath = _localPathFromUrlOrPath(urlOrPath)
+    if localPath is not None:
+        return (localPath, 'local')
+    log.info('Retrieving %s to %s.' % (urlOrPath, destPath))
+    _downloadWithFallback(urlOrPath, destPath)
+    return (destPath, 'remote')
+
+
+def _localPathFromUrlOrPath(urlOrPath):
+    '''
+    Returns the local filesystem path for a plain path or a ``file://`` URL, or ``None``
+    when ``urlOrPath`` points at a remote resource.
+    '''
     urlp = urlparse(urlOrPath)
-    if urlp.scheme == '':
-        return (os.path.abspath(urlOrPath), 'local')
+    if urlp.scheme == 'file':
+        return os.path.abspath(url2pathname(urlp.path))
+    if len(urlp.scheme) <= 1:
+        # no scheme, or a single letter which is a Windows drive such as C:\...
+        return os.path.abspath(urlOrPath)
+    return None
+
+
+def _fallbackDistributionUrl(url):
+    '''
+    Maps a URL on the Apache dist mirrors (which only carry current releases) to the
+    permanent copy on archive.apache.org. Returns ``None`` for any other URL.
+    '''
+    urlp = urlparse(url)
+    if urlp.scheme in ('http', 'https') and urlp.hostname in ('dlcdn.apache.org', 'downloads.apache.org') \
+            and urlp.path.startswith('/tika/'):
+        return TikaArchiveBaseUrl + urlp.path[len('/tika/'):]
+    return None
+
+
+def _downloadWithFallback(url, filename):
+    '''
+    Downloads ``url`` to ``filename``. When an Apache dist mirror no longer carries the
+    requested release (404) the download is retried once from archive.apache.org.
+    '''
+    try:
+        return _urlretrieve(url, filename)
+    except TikaDownloadError as e:
+        fallback = _fallbackDistributionUrl(url)
+        if e.status_code != 404 or fallback is None:
+            raise
+        log.warning('%s not found (404); retrying from %s', url, fallback)
+        return _urlretrieve(fallback, filename)
+
+
+def _parseChecksumFile(text, algo):
+    '''
+    Extracts the hex digest from a checksum sidecar. Accepts a bare digest, the
+    ``<digest>  <filename>`` form and the ``gpg --print-md`` block form.
+    :return: lowercase hex digest
+    '''
+    expected = hashlib.new(algo).digest_size * 2
+    for token in text.split():
+        if len(token) == expected and re.fullmatch(r'[0-9a-fA-F]+', token):
+            return token.lower()
+    # gpg --print-md style: "<filename>: AB12 CD34 ..." possibly wrapped over several lines
+    compact = re.sub(r'[^0-9a-fA-F]', '', text.split(':', 1)[-1])
+    if len(compact) == expected:
+        return compact.lower()
+    raise TikaException('No %s digest found in checksum file' % algo)
+
+
+def _fileDigest(path, algo, chunk_size=1 << 20):
+    '''
+    Hex digest of a file, computed in chunks.
+    '''
+    m = hashlib.new(algo)
+    with open(path, 'rb') as f:
+        for chunk in iter(lambda: f.read(chunk_size), b''):
+            m.update(chunk)
+    return m.hexdigest()
+
+
+def _findServerJar(installDir):
+    '''
+    Returns the path of the Tika server launcher jar inside an extracted distribution,
+    or ``None`` if ``installDir`` does not hold a complete distribution (jar plus ``lib/``).
+    '''
+    if not os.path.isdir(os.path.join(installDir, 'lib')):
+        return None
+    jars = [n for n in os.listdir(installDir) if n.lower().endswith('.jar')]
+    preferred = sorted(n for n in jars if n.startswith('tika-server-standard-'))
+    if preferred:
+        return os.path.join(installDir, preferred[0])
+    if len(jars) == 1:
+        return os.path.join(installDir, jars[0])
+    return None
+
+
+def _extractServerDistribution(archivePath, installDir):
+    '''
+    Extracts the Tika server zip distribution into ``installDir``. The archive is unpacked
+    into a temporary sibling directory that is renamed into place at the end, so
+    ``installDir`` only ever exists once it is complete.
+    :return: ``installDir``
+    '''
+    installDir = os.path.abspath(installDir)
+    parent = os.path.dirname(installDir)
+    os.makedirs(parent, exist_ok=True)
+    tmpDir = tempfile.mkdtemp(prefix=os.path.basename(installDir) + '.', dir=parent)
+    try:
+        with zipfile.ZipFile(archivePath) as archive:
+            archive.extractall(tmpDir)
+        # tolerate a distribution wrapped in a single top-level directory
+        source = tmpDir
+        entries = os.listdir(tmpDir)
+        if len(entries) == 1 and _findServerJar(os.path.join(tmpDir, entries[0])):
+            source = os.path.join(tmpDir, entries[0])
+        try:
+            os.rename(source, installDir)
+        except OSError:
+            if not _findServerJar(installDir):
+                raise
+            # another process extracted the same distribution first; keep theirs
+    finally:
+        shutil.rmtree(tmpDir, ignore_errors=True)
+    return installDir
+
+
+def _javaMajorVersion(java_path=TikaJava):
+    '''
+    Major version of the given Java runtime (8 for 1.8.x, 17 for 17.x), or ``None`` when
+    the binary cannot be run or its version output cannot be parsed.
+    '''
+    try:
+        proc = subprocess.run([java_path, '-version'], capture_output=True, text=True, timeout=30)
+    except (OSError, subprocess.SubprocessError):
+        return None
+    match = re.search(r'version "(\d+)(?:\.(\d+))?', proc.stderr + proc.stdout)
+    if not match:
+        return None
+    major = int(match.group(1))
+    if major == 1 and match.group(2):
+        return int(match.group(2))
+    return major
+
+
+def _jarNeedsLibDir(jarPath):
+    '''
+    ``True`` when the jar's manifest references a ``lib/`` class path, i.e. it is a thin
+    launcher that only runs next to the ``lib/`` directory of its distribution.
+    '''
+    try:
+        with zipfile.ZipFile(jarPath) as jar:
+            manifest = jar.read('META-INF/MANIFEST.MF').decode('utf-8', 'replace')
+    except (OSError, KeyError, zipfile.BadZipFile):
+        return False
+    manifest = manifest.replace('\r\n ', '').replace('\n ', '')  # unwrap continuation lines
+    return bool(re.search(r'^Class-Path:.*\blib/', manifest, re.MULTILINE))
+
+
+def _removeQuietly(*paths):
+    for path in paths:
+        try:
+            os.remove(path)
+        except OSError:
+            pass
+
+
+def ensureTikaServerJar(tikaServerJar=TikaServerJar):
+    '''
+    Makes sure a runnable Tika server is available locally and returns the path of its
+    launcher jar.
+
+    ``tikaServerJar`` is normally the URL of the ``tika-server-standard-<version>.zip``
+    distribution. It may also be a ``file://`` URL or plain path to a local copy of that
+    zip, or a URL/path ending in ``.jar`` for a jar that already sits next to its ``lib/``
+    directory (or a self-contained pre-4.0 jar). Remote downloads are verified against
+    their ``.<TIKA_JAR_HASH_ALGO>`` sidecar; a zip is extracted once into
+    ``<TIKA_PATH>/<zip name without extension>/`` and reused on later runs.
+    :return: path of the server jar to launch
+    '''
+    sourcePath = _localPathFromUrlOrPath(tikaServerJar)
+    name = os.path.basename(sourcePath if sourcePath is not None else urlparse(tikaServerJar).path)
+    if not name:
+        raise TikaException('Cannot determine the file name of the Tika server distribution from %r' % (tikaServerJar,))
+    checksumSuffix = '.' + TikaJarHashAlgo
+
+    if name.lower().endswith('.jar'):
+        if sourcePath is not None:
+            if not os.path.isfile(sourcePath):
+                raise TikaException('Tika server jar %s does not exist' % sourcePath)
+            jarPath = sourcePath
+        else:
+            jarPath = os.path.join(TikaJarPath, name)
+            if not os.path.isfile(jarPath):
+                getRemoteJar(tikaServerJar, jarPath)
+            if not checkJarSig(tikaServerJar, jarPath):
+                log.warning('Checksum mismatch for %s; downloading again.', jarPath)
+                _removeQuietly(jarPath, jarPath + checksumSuffix)
+                getRemoteJar(tikaServerJar, jarPath)
+                if not checkJarSig(tikaServerJar, jarPath):
+                    raise TikaException('Checksum mismatch for %s after re-download' % jarPath)
+        if _jarNeedsLibDir(jarPath) and not os.path.isdir(os.path.join(os.path.dirname(jarPath), 'lib')):
+            log.warning('%s is a thin launcher without a lib/ directory next to it; since Apache Tika 4 '
+                        'the server only runs from inside the unzipped tika-server-standard distribution.', jarPath)
+        return jarPath
+
+    stem = name[:-4] if name.lower().endswith('.zip') else name
+    installDir = os.path.join(TikaJarPath, stem)
+    jarPath = _findServerJar(installDir)
+    if jarPath:
+        return jarPath
+
+    if sourcePath is not None:
+        archivePath = sourcePath
+        if not os.path.isfile(archivePath):
+            raise TikaException('Tika server distribution %s does not exist' % archivePath)
+        if not checkJarSig(tikaServerJar, archivePath):
+            raise TikaException('Checksum mismatch for %s' % archivePath)
     else:
-        log.info('Retrieving %s to %s.' % (urlOrPath, destPath))
-        _urlretrieve(urlOrPath, destPath)
-        return (destPath, 'remote')
+        archivePath = os.path.join(TikaJarPath, name)
+        if not os.path.isfile(archivePath):
+            getRemoteJar(tikaServerJar, archivePath)
+        if not checkJarSig(tikaServerJar, archivePath):
+            log.warning('Checksum mismatch for %s; downloading again.', archivePath)
+            _removeQuietly(archivePath, archivePath + checksumSuffix)
+            getRemoteJar(tikaServerJar, archivePath)
+            if not checkJarSig(tikaServerJar, archivePath):
+                raise TikaException('Checksum mismatch for %s after re-download' % archivePath)
+
+    log.info('Extracting %s to %s.', archivePath, installDir)
+    _extractServerDistribution(archivePath, installDir)
+    jarPath = _findServerJar(installDir)
+    if not jarPath:
+        raise TikaException('No tika-server-standard-*.jar with a lib/ directory found in %s' % installDir)
+    return jarPath
 
 def checkPortIsOpen(remoteServerHost=ServerHost, port = Port):
     '''
diff --git a/tika/translate.py b/tika/translate.py
deleted file mode 100644
index 7ddb8bb8..00000000
--- a/tika/translate.py
+++ /dev/null
@@ -1,66 +0,0 @@
-# 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.
-#
-
-from .tika import ServerEndpoint, Translator, callServer, doTranslate1
-
-
-def from_file(filename, srcLang, destLang, serverEndpoint=ServerEndpoint, requestOptions={}):
-    '''
-    Traslates the content of source file to destination language
-    :param filename: file whose contents needs translation
-    :param srcLang: name of language of input file
-    :param destLang: name of language of desired language
-    :param serverEndpoint: Tika server end point (Optional)
-    :return: translated content
-    '''
-    jsonOutput = doTranslate1(srcLang+':'+destLang, filename, serverEndpoint, requestOptions=requestOptions)
-    return jsonOutput[1]
-
-def from_buffer(string, srcLang, destLang, serverEndpoint=ServerEndpoint, requestOptions={}):
-    '''
-    Translates content from source language to desired destination language
-    :param string: input content which needs translation
-    :param srcLang: name of language of the input content
-    :param destLang: name of the desired language for translation
-    :param serverEndpoint:
-    :return:
-    '''
-    status, response = callServer('put', ServerEndpoint, '/translate/all/'+Translator+'/'+srcLang+'/'+destLang,
-                                  string, {'Accept': 'text/plain'}, False, requestOptions=requestOptions)
-    return response
-
-def auto_from_file(filename, destLang, serverEndpoint=ServerEndpoint, requestOptions={}):
-    '''
-    Translates contents of a file to desired language by auto detecting the source language
-    :param filename: file whose contents needs translation
-    :param destLang: name of the desired language for translation
-    :param serverEndpoint: Tika server end point (Optional)
-    :return:
-    '''
-    jsonOutput = doTranslate1(destLang, filename, serverEndpoint, requestOptions=requestOptions)
-    return jsonOutput[1]
-
-def auto_from_buffer(string, destLang, serverEndpoint=ServerEndpoint, requestOptions={}):
-    '''
-    Translates content to desired language by auto detecting the source language
-    :param string: input content which needs translation
-    :param destLang: name of the desired language for translation
-    :param serverEndpoint: Tika server end point (Optional)
-    :return:
-    '''
-    status, response = callServer('put', ServerEndpoint, '/translate/all/'+Translator+'/'+destLang,
-                                  string, {'Accept': 'text/plain'}, False, requestOptions=requestOptions)
-    return response
diff --git a/tika/unpack.py b/tika/unpack.py
index 0850e7af..116a0d15 100644
--- a/tika/unpack.py
+++ b/tika/unpack.py
@@ -14,29 +14,38 @@
 # limitations under the License.
 #
 
-import csv
-import tarfile
-from contextlib import closing
-from io import BytesIO, TextIOWrapper
+import json
+import re
+import zipfile
+from io import BytesIO
 
 from .tika import ServerEndpoint, callServer, parse1
 
-_text_wrapper = TextIOWrapper
+# Tika 4 /unpack/all output: a zip holding the original document as ``0.<ext>``, embedded
+# documents as ``1.<ext>``, ``2.<ext>``, ... and one ``<name>.metadata.json`` sidecar per entry.
+# The trailing handler name selects plain text for the extracted content (default: Markdown).
+UNPACK_SERVICE = '/unpack/all/text'
+_SIDECAR_SUFFIX = '.metadata.json'
+_CONTENT_KEY = 'tk:content'
 
 
 def from_file(filename, serverEndpoint=ServerEndpoint, requestOptions={}):
     '''
-    Parse from file
+    Parse from file and unpack the documents embedded in it.
+
+    Returns a dict with ``content`` and ``metadata`` of the document and ``attachments``
+    mapping the name of each embedded document to its bytes, or ``{}`` when the server
+    returned nothing.
+
     :param filename: file
     :param serverEndpoint: Tika server end point (optional)
-    :return:
     '''
-    tarOutput = parse1('unpack', filename, serverEndpoint,
-                       responseMimeType='application/x-tar',
-                       services={'meta': '/meta', 'text': '/tika',
-                                 'all': '/rmeta/xml', 'unpack': '/unpack/all'},
+    zipOutput = parse1('unpack', filename, serverEndpoint,
+                       responseMimeType='application/zip',
+                       services={'meta': '/meta', 'text': '/tika/text',
+                                 'all': '/rmeta/xml', 'unpack': UNPACK_SERVICE},
                        rawResponse=True, requestOptions=requestOptions)
-    return _parse(tarOutput)
+    return _parse(zipOutput)
 
 
 def from_buffer(string, serverEndpoint=ServerEndpoint, headers=None, requestOptions={}):
@@ -44,77 +53,63 @@ def from_buffer(string, serverEndpoint=ServerEndpoint, headers=None, requestOpti
     Parse from buffered content
     :param string:  buffered content
     :param serverEndpoint: Tika server URL (Optional)
-    :return: parsed content
+    :return: parsed content, see ``from_file``
     '''
 
     headers = headers or {}
-    headers.update({'Accept': 'application/x-tar'})
+    headers.update({'Accept': 'application/zip'})
 
-    status, response = callServer('put', serverEndpoint, '/unpack/all', string,
+    status, response = callServer('put', serverEndpoint, UNPACK_SERVICE, string,
                                   headers, False,
                                   rawResponse=True, requestOptions=requestOptions)
 
     return _parse((status, response))
 
 
-def _parse(tarOutput):
-    parsed = {}
-    if not tarOutput:
-        return parsed
-    elif tarOutput[1] is None or tarOutput[1] == b"":
-        return parsed
+def _parse(zipOutput):
+    '''
+    Turns a ``/unpack/all`` zip response into ``content`` and ``metadata`` of the container
+    document (taken from its ``.metadata.json`` sidecar) and ``attachments`` mapping the
+    name of each embedded document to its bytes. Sidecars and the container itself are
+    not attachments.
+    :param zipOutput: ``(status, bytes)`` tuple as returned by ``callServer(rawResponse=True)``
+    :return: ``{}`` for an empty response, else a dict with ``content``, ``metadata``, ``attachments``
+    '''
+    if not zipOutput or not zipOutput[1]:
+        return {}
+
+    with zipfile.ZipFile(BytesIO(zipOutput[1])) as archive:
+        names = [info.filename for info in archive.infolist() if not info.is_dir()]
+        sidecars = {n for n in names if n.endswith(_SIDECAR_SUFFIX)}
+        files = sorted(n for n in names if n not in sidecars)
 
-    with tarfile.open(fileobj=BytesIO(tarOutput[1])) as tarFile:
-        # get the member names
-        memberNames = list(tarFile.getnames())
+        container = next((n for n in files if re.match(r'0(\.|$)', n)), files[0] if files else None)
 
-        # extract the metadata
         metadata = {}
-        if "__METADATA__" in memberNames:
-            memberNames.remove("__METADATA__")
-
-        metadataMember = tarFile.getmember("__METADATA__")
-        if not metadataMember.issym() and metadataMember.isfile():
-            with closing(_text_wrapper(tarFile.extractfile(metadataMember), encoding=tarFile.encoding)) as metadataFile:
-                metadataReader = csv.reader(_truncate_nulls(metadataFile))
-                for metadataLine in metadataReader:
-                    # each metadata line comes as a key-value pair, with list values
-                    # returned as extra values in the line - convert single values
-                    # to non-list values to be consistent with parser metadata
-                    assert len(metadataLine) >= 2
-
-                    if len(metadataLine) > 2:
-                        metadata[metadataLine[0]] = metadataLine[1:]
-                    else:
-                        metadata[metadataLine[0]] = metadataLine[1]
-
-
-        # get the content
-        content = ""
-        if "__TEXT__" in memberNames:
-            memberNames.remove("__TEXT__")
-
-            contentMember = tarFile.getmember("__TEXT__")
-            if not contentMember.issym() and contentMember.isfile():
-                with closing(_text_wrapper(tarFile.extractfile(contentMember), encoding='utf8')) as content_file:
-                    content = content_file.read()
-
-        # get the remaining files as attachments
-        attachments = {}
-        for attachment in memberNames:
-            attachmentMember = tarFile.getmember(attachment)
-            if not attachmentMember.issym() and attachmentMember.isfile():
-                with closing(tarFile.extractfile(attachmentMember)) as attachment_file:
-                    attachments[attachment] = attachment_file.read()
-
-        parsed["content"] = content
-        parsed["metadata"] = metadata
-        parsed["attachments"] = attachments
-
-        return parsed
-
-
-# TODO: Remove if/when fixed. https://issues.apache.org/jira/browse/TIKA-3070
-def _truncate_nulls(s):
-    for line in s:
-        yield line.replace('\0', '')
+        for candidate in _sidecar_candidates(container, sidecars):
+            metadata = json.loads(archive.read(candidate).decode('utf-8'))
+            break
+        if isinstance(metadata, list):  # tolerate an rmeta style list; the container comes first
+            metadata = metadata[0] if metadata else {}
+
+        content = metadata.pop(_CONTENT_KEY, '')
+        if isinstance(content, list):
+            content = ''.join(content)
+
+        attachments = {n: archive.read(n) for n in files if n != container}
+
+    return {'content': content, 'metadata': metadata, 'attachments': attachments}
+
+
+def _sidecar_candidates(container, sidecars):
+    '''
+    Sidecar names that may describe the container document, most specific first:
+    ``0.pdf.metadata.json``, then ``0.metadata.json``.
+    '''
+    candidates = []
+    if container is not None:
+        candidates.append(container + _SIDECAR_SUFFIX)
+        stem = container.rsplit('.', 1)[0]
+        candidates.append(stem + _SIDECAR_SUFFIX)
+    candidates.append('0' + _SIDECAR_SUFFIX)
+    return [c for c in candidates if c in sidecars]