Skip to content

Remove lombok e2e #2925

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
tballison wants to merge 4 commits into
base: main
Choose a base branch
from
+446 −224
Open

Remove lombok e2e #2925

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
4d91c53
TIKA-4764 and TIKA-4763 merge conflicts
tballison Jun 26, 2026
493814a
TIKA-476x - fix merge conflicts, simplify tika.proto
tballison Jun 26, 2026
2c07998
TIKA-4763 fix e2e
tballison Jun 26, 2026
53bbcb9
follow on to TIKA-4763 fix e2e
tballison Jul 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 14 additions & 10 deletions docs/modules/ROOT/pages/using-tika/grpc/index.adoc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,13 +67,16 @@ the server's defaults for that request. Because this can reconfigure any pipelin
component (fetcher, parser, timeouts, ...), it is off by default. When `false`, a
request carrying either field is rejected with `PERMISSION_DENIED`.

|`allowComponentModifications`
|When `true`, callers may add, modify, or delete fetchers and pipes iterators at
runtime (`SaveFetcher`, `DeleteFetcher`, `SavePipesIterator`,
`DeletePipesIterator`). Because this changes what the server can reach for all
subsequent requests (for example, adding a fetcher that escapes a configured base
path), it is off by default. When `false`, those RPCs are rejected with
`PERMISSION_DENIED`.
|`allowComponentManagement`
|When `true`, callers may manage fetchers and pipes iterators at runtime: add,
modify, or delete them (`SaveFetcher`, `DeleteFetcher`, `SavePipesIterator`,
`DeletePipesIterator`), and read their stored configuration back (`GetFetcher`,
`ListFetchers`, `GetPipesIterator`). It is off by default for two reasons:
mutations change what the server can reach for all subsequent requests (for
example, adding a fetcher that escapes a configured base path), and the stored
configs returned by the read RPCs may contain secrets (passwords, access keys,
tokens). When `false`, the mutating RPCs are rejected with `PERMISSION_DENIED`,
and the read RPCs return only component identity (id and class), never the config.
|===

Enable these only for trusted callers over a secured channel:
Expand All @@ -83,7 +86,7 @@ Enable these only for trusted callers over a secured channel:
{
"grpc": {
"allowPerRequestConfig": true,
"allowComponentModifications": true
"allowComponentManagement": true
}
}
----
Expand Down Expand Up @@ -149,8 +152,9 @@ automatically. Two distinct controls are involved, and both matter:
Mesh mTLS authenticates *who opened the connection*; it does not authorize *what
that caller may do*, so an authenticated-but-untrusted pod can still invoke
whatever RPC surface is enabled — at minimum `FetchAndParse` against your
configured fetchers, and the runtime-mutation RPCs too if you have set
`allowComponentModifications`. Restricting reachability with a `NetworkPolicy` is
configured fetchers, and the runtime component-management RPCs too (including the
config reads that can expose stored secrets) if you have set
`allowComponentManagement`. Restricting reachability with a `NetworkPolicy` is
therefore required, not optional — running in Kubernetes without one leaves
tika-grpc reachable by every pod in the cluster.

Expand Down
8 changes: 0 additions & 8 deletions tika-e2e-tests/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -103,14 +103,6 @@
<artifactId>jackson-databind</artifactId>
<version>${jackson.version}</version>
</dependency>

<!-- Lombok -->
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>${lombok.version}</version>
<optional>true</optional>
</dependency>
</dependencies>
</dependencyManagement>

Expand Down
7 changes: 0 additions & 7 deletions tika-e2e-tests/tika-grpc/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,13 +70,6 @@
<artifactId>jackson-databind</artifactId>
</dependency>

<!-- Lombok -->
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>

<!-- JUnit 5 -->
<dependency>
<groupId>org.junit.jupiter</groupId>
Expand Down
43 changes: 22 additions & 21 deletions tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/ExternalTestBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,12 +43,13 @@
import com.fasterxml.jackson.databind.ObjectMapper;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.TestInstance;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testcontainers.containers.DockerComposeContainer;
import org.testcontainers.containers.output.Slf4jLogConsumer;
import org.testcontainers.containers.wait.strategy.Wait;
Expand All @@ -60,9 +61,9 @@

@TestInstance(TestInstance.Lifecycle.PER_CLASS)
@Testcontainers
@Slf4j
@Tag("E2ETest")
public abstract class ExternalTestBase {
private static final Logger LOG = LoggerFactory.getLogger(ExternalTestBase.class);
public static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
public static final int MAX_STARTUP_TIMEOUT = 120;
public static final String GOV_DOCS_FOLDER = "/tika/govdocs1";
Expand All @@ -88,7 +89,7 @@ static void setup() throws Exception {
}

private static void startLocalGrpcServer() throws Exception {
log.info("Starting local tika-grpc server using Maven exec");
LOG.info("Starting local tika-grpc server using Maven exec");

Path tikaGrpcDir = findTikaGrpcDirectory();
Path configFile = Path.of("src/test/resources/tika-config.json").toAbsolutePath();
Expand All @@ -97,8 +98,8 @@ private static void startLocalGrpcServer() throws Exception {
throw new IllegalStateException("Config file not found: " + configFile);
}

log.info("Using tika-grpc from: {}", tikaGrpcDir);
log.info("Using config file: {}", configFile);
LOG.info("Using tika-grpc from: {}", tikaGrpcDir);
LOG.info("Using config file: {}", configFile);

String javaHome = System.getProperty("java.home");
boolean isWindows = System.getProperty("os.name").toLowerCase(Locale.ROOT).contains("win");
Expand Down Expand Up @@ -131,18 +132,18 @@ private static void startLocalGrpcServer() throws Exception {
new InputStreamReader(localGrpcProcess.getInputStream(), StandardCharsets.UTF_8))) {
String line;
while ((line = reader.readLine()) != null) {
log.info("tika-grpc: {}", line);
LOG.info("tika-grpc: {}", line);
}
} catch (IOException e) {
log.error("Error reading server output", e);
LOG.error("Error reading server output", e);
}
});
logThread.setDaemon(true);
logThread.start();

waitForServerReady();

log.info("Local tika-grpc server started successfully on port {}", GRPC_PORT);
LOG.info("Local tika-grpc server started successfully on port {}", GRPC_PORT);
}

private static Path findTikaGrpcDirectory() {
Expand Down Expand Up @@ -173,10 +174,10 @@ private static void waitForServerReady() throws Exception {
try {
TikaGrpc.TikaBlockingStub stub = TikaGrpc.newBlockingStub(testChannel);
stub.listFetchers(ListFetchersRequest.newBuilder().build());
log.info("gRPC server is ready");
LOG.info("gRPC server is ready");
return;
} catch (Exception e) {
log.trace("gRPC server not ready yet (attempt {}/{}): {}", i + 1, maxAttempts, e.getMessage());
LOG.trace("gRPC server not ready yet (attempt {}/{}): {}", i + 1, maxAttempts, e.getMessage());
} finally {
testChannel.shutdown();
testChannel.awaitTermination(1, TimeUnit.SECONDS);
Expand All @@ -191,7 +192,7 @@ private static void waitForServerReady() throws Exception {
}

private static void startDockerGrpcServer() {
log.info("Starting Docker Compose tika-grpc server");
LOG.info("Starting Docker Compose tika-grpc server");

String composeFilePath = System.getProperty("tika.docker.compose.file");
if (composeFilePath == null || composeFilePath.isBlank()) {
Expand All @@ -208,11 +209,11 @@ private static void startDockerGrpcServer() {
.withStartupTimeout(Duration.of(MAX_STARTUP_TIMEOUT, ChronoUnit.SECONDS))
.withExposedService("tika-grpc", 50052,
Wait.forLogMessage(".*Server started.*\\n", 1))
.withLogConsumer("tika-grpc", new Slf4jLogConsumer(log));
.withLogConsumer("tika-grpc", new Slf4jLogConsumer(LOG));

composeContainer.start();

log.info("Docker Compose containers started successfully");
LOG.info("Docker Compose containers started successfully");
}

private static void loadGovdocs1() throws IOException, InterruptedException {
Expand All @@ -230,7 +231,7 @@ private static void loadGovdocs1() throws IOException, InterruptedException {
if (attempt >= retries) {
throw e;
}
log.warn("Download attempt {} failed, retrying in 10 seconds...", attempt, e);
LOG.warn("Download attempt {} failed, retrying in 10 seconds...", attempt, e);
TimeUnit.SECONDS.sleep(10);
}
}
Expand All @@ -253,13 +254,13 @@ public static void copyTestFixtures() throws IOException {
Files.copy(in, targetDir.resolve(fixture), StandardCopyOption.REPLACE_EXISTING);
}
}
log.info("Copied {} test fixtures to {}", fixtures.length, targetDir);
LOG.info("Copied {} test fixtures to {}", fixtures.length, targetDir);
}

@AfterAll
void close() {
if (USE_LOCAL_SERVER && localGrpcProcess != null) {
log.info("Stopping local gRPC server");
LOG.info("Stopping local gRPC server");
localGrpcProcess.destroy();
try {
if (!localGrpcProcess.waitFor(10, TimeUnit.SECONDS)) {
Expand All @@ -284,14 +285,14 @@ public static void downloadAndUnzipGovdocs1(int fromIndex, int toIndex) throws I
Path zipPath = targetDir.resolve(zipName);

if (Files.exists(zipPath)) {
log.info("{} already exists, skipping download", zipName);
LOG.info("{} already exists, skipping download", zipName);
} else {
log.info("Downloading {} from {}...", zipName, url);
LOG.info("Downloading {} from {}...", zipName, url);
try (InputStream in = new URL(url).openStream()) {
Files.copy(in, zipPath, StandardCopyOption.REPLACE_EXISTING);
}
}
log.info("Unzipping {}...", zipName);
LOG.info("Unzipping {}...", zipName);
try (ZipInputStream zis = new ZipInputStream(new FileInputStream(zipPath.toFile()))) {
ZipEntry entry;
while ((entry = zis.getNextEntry()) != null) {
Expand All @@ -309,7 +310,7 @@ public static void downloadAndUnzipGovdocs1(int fromIndex, int toIndex) throws I
}
}

log.info("Finished downloading and extracting govdocs1 files");
LOG.info("Finished downloading and extracting govdocs1 files");
}

public static void assertAllFilesFetched(Path baseDir, List<FetchAndParseReply> successes,
Expand Down Expand Up @@ -337,7 +338,7 @@ public static void assertAllFilesFetched(Path baseDir, List<FetchAndParseReply>
}

Assertions.assertNotEquals(0, successes.size(), "Should have some successful fetches");
log.info("Processed {} files: {} successes, {} errors", allFetchKeys.size(), successes.size(), errors.size());
LOG.info("Processed {} files: {} successes, {} errors", allFetchKeys.size(), successes.size(), errors.size());
Assertions.assertEquals(keysFromGovdocs1, allFetchKeys, () -> {
Set<String> missing = new HashSet<>(keysFromGovdocs1);
missing.removeAll(allFetchKeys);
Expand Down
26 changes: 14 additions & 12 deletions ...tests/tika-grpc/src/test/java/org/apache/tika/pipes/filesystem/FileSystemFetcherTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,11 +27,12 @@

import io.grpc.ManagedChannel;
import io.grpc.stub.StreamObserver;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.condition.DisabledOnOs;
import org.junit.jupiter.api.condition.OS;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.tika.FetchAndParseReply;
import org.apache.tika.FetchAndParseRequest;
Expand All @@ -41,9 +42,10 @@
import org.apache.tika.pipes.ExternalTestBase;
import org.apache.tika.pipes.fetcher.fs.FileSystemFetcherConfig;

@Slf4j
@DisabledOnOs(value = OS.WINDOWS, disabledReason = "exec:exec classpath exceeds Windows CreateProcess command-line length limit")
class FileSystemFetcherTest extends ExternalTestBase {
private static final Logger LOG = LoggerFactory.getLogger(FileSystemFetcherTest.class);


@Test
void testFileSystemFetcher() throws Exception {
Expand All @@ -59,16 +61,16 @@ void testFileSystemFetcher() throws Exception {
config.setBasePath(basePath);

String configJson = OBJECT_MAPPER.writeValueAsString(config);
log.info("Creating fetcher with config (basePath={}): {}", basePath, configJson);
LOG.info("Creating fetcher with config (basePath={}): {}", basePath, configJson);

SaveFetcherReply saveReply = blockingStub.saveFetcher(SaveFetcherRequest
.newBuilder()
.setFetcherId(fetcherId)
.setFetcherClass("org.apache.tika.pipes.fetcher.fs.FileSystemFetcher")
.setFetcherType("file-system-fetcher")
.setFetcherConfigJson(configJson)
.build());

log.info("Fetcher created: {}", saveReply.getFetcherId());
LOG.info("Fetcher created: {}", saveReply.getFetcherId());

List<FetchAndParseReply> successes = Collections.synchronizedList(new ArrayList<>());
List<FetchAndParseReply> errors = Collections.synchronizedList(new ArrayList<>());
Expand All @@ -78,7 +80,7 @@ void testFileSystemFetcher() throws Exception {
requestStreamObserver = tikaStub.fetchAndParseBiDirectionalStreaming(new StreamObserver<>() {
@Override
public void onNext(FetchAndParseReply fetchAndParseReply) {
log.debug("Reply from fetch-and-parse - key={}, status={}",
LOG.debug("Reply from fetch-and-parse - key={}, status={}",
fetchAndParseReply.getFetchKey(), fetchAndParseReply.getStatus());
if ("FETCH_AND_PARSE_EXCEPTION".equals(fetchAndParseReply.getStatus())) {
errors.add(fetchAndParseReply);
Expand All @@ -89,14 +91,14 @@ public void onNext(FetchAndParseReply fetchAndParseReply) {

@Override
public void onError(Throwable throwable) {
log.error("Received an error", throwable);
LOG.error("Received an error", throwable);
Assertions.fail(throwable);
countDownLatch.countDown();
}

@Override
public void onCompleted() {
log.info("Finished streaming fetch and parse replies");
LOG.info("Finished streaming fetch and parse replies");
countDownLatch.countDown();
}
});
Expand All @@ -122,13 +124,13 @@ public void onCompleted() {
}
});
}
log.info("Done submitting files to fetcher {}", fetcherId);
LOG.info("Done submitting files to fetcher {}", fetcherId);

requestStreamObserver.onCompleted();

try {
if (!countDownLatch.await(3, TimeUnit.MINUTES)) {
log.error("Timed out waiting for parse to complete");
LOG.error("Timed out waiting for parse to complete");
Assertions.fail("Timed out waiting for parsing to complete");
}
} catch (InterruptedException e) {
Expand All @@ -140,14 +142,14 @@ public void onCompleted() {
assertAllFilesFetched(TEST_FOLDER.toPath(), successes, errors);
} else {
int totalProcessed = successes.size() + errors.size();
log.info("Processed {} documents (limit was {})", totalProcessed, maxDocs);
LOG.info("Processed {} documents (limit was {})", totalProcessed, maxDocs);
Assertions.assertTrue(totalProcessed <= maxDocs,
"Should not process more than " + maxDocs + " documents");
Assertions.assertTrue(totalProcessed > 0,
"Should have processed at least one document");
}

log.info("Test completed successfully - {} successes, {} errors",
LOG.info("Test completed successfully - {} successes, {} errors",
successes.size(), errors.size());
} finally {
channel.shutdown();
Expand Down
Loading