This is an automated email from the ASF dual-hosted git repository. dominikriemer pushed a commit to branch remove-lightcouch-dependency in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit fed90de9d6a171c1ead50eccce8f8eaf8bd5b31b Author: Dominik Riemer <[email protected]> AuthorDate: Tue Nov 3 11:36:08 2026 +0100 Fix database creation --- .../streampipes/couchdb/client/CouchDatabase.java | 18 ++++++------ .../streampipes/couchdb/client/CouchServer.java | 23 ++++++++++++--- .../client/CouchDbClientIntegrationTest.java | 34 ++++++++++++++++++++++ .../couchdb/client/TypedCollectionTest.java | 31 +++++++++++++++++--- .../streampipes/dataexplorer/ImageStore.java | 2 +- .../streampipes/dataexplorer/ImageStoreTest.java | 26 +++++++++++------ .../sinks/databases/jvm/couchdb/CouchDbSink.java | 2 +- .../databases/jvm/couchdb/CouchDbSinkTest.java | 20 ++++++++----- 8 files changed, 120 insertions(+), 36 deletions(-) diff --git a/streampipes-couchdb-client/src/main/java/org/apache/streampipes/couchdb/client/CouchDatabase.java b/streampipes-couchdb-client/src/main/java/org/apache/streampipes/couchdb/client/CouchDatabase.java index 2bb45373c6..5ace93eb1b 100644 --- a/streampipes-couchdb-client/src/main/java/org/apache/streampipes/couchdb/client/CouchDatabase.java +++ b/streampipes-couchdb-client/src/main/java/org/apache/streampipes/couchdb/client/CouchDatabase.java @@ -20,7 +20,6 @@ package org.apache.streampipes.couchdb.client; import org.apache.streampipes.couchdb.client.codec.DocumentCodec; import org.apache.streampipes.couchdb.client.codec.JacksonCodec; -import org.apache.streampipes.couchdb.client.exception.CouchDbException; import org.apache.streampipes.couchdb.client.exception.DatabaseNotFoundException; import org.apache.streampipes.couchdb.client.exception.PreconditionFailedException; import org.apache.streampipes.couchdb.client.transport.CouchRequest; @@ -59,14 +58,7 @@ public final class CouchDatabase { } public boolean exists() { - try { - return server.execute(CouchRequest.head(path), response -> true); - } catch (CouchDbException e) { - if (e.getStatus() == 404) { - return false; - } - throw e; - } + return server.databaseExists(name); } /** @@ -83,6 +75,14 @@ public final class CouchDatabase { } } + /** + * Creates the database unless it exists. A user who may use an existing database but not create databases can call + * this, unlike {@link #create()}. + */ + public void ensureExists() { + server.ensureDatabase(name); + } + public void delete() { server.forgetDatabase(name); server.execute(CouchRequest.delete(path), response -> null); diff --git a/streampipes-couchdb-client/src/main/java/org/apache/streampipes/couchdb/client/CouchServer.java b/streampipes-couchdb-client/src/main/java/org/apache/streampipes/couchdb/client/CouchServer.java index 5746ec62d8..35014a7899 100644 --- a/streampipes-couchdb-client/src/main/java/org/apache/streampipes/couchdb/client/CouchServer.java +++ b/streampipes-couchdb-client/src/main/java/org/apache/streampipes/couchdb/client/CouchServer.java @@ -18,6 +18,7 @@ package org.apache.streampipes.couchdb.client; +import org.apache.streampipes.couchdb.client.exception.CouchDbException; import org.apache.streampipes.couchdb.client.exception.CouchDbTransportException; import org.apache.streampipes.couchdb.client.exception.DocumentMappingException; import org.apache.streampipes.couchdb.client.exception.PreconditionFailedException; @@ -139,14 +140,28 @@ public final class CouchServer implements AutoCloseable { if (knownDatabases.contains(name)) { return; } - try { - execute(new CouchRequest("PUT", CouchPaths.database(name), CouchRequest.JSON, null, null), response -> null); - } catch (PreconditionFailedException e) { - // 412: the database exists already. + // Checked first because creating needs admin rights that using an existing database does not. + if (!databaseExists(name)) { + try { + execute(new CouchRequest("PUT", CouchPaths.database(name), CouchRequest.JSON, null, null), response -> null); + } catch (PreconditionFailedException e) { + // 412: created by someone else since the check. + } } knownDatabases.add(name); } + boolean databaseExists(String name) { + try { + return execute(CouchRequest.head(CouchPaths.database(name)), response -> true); + } catch (CouchDbException e) { + if (e.getStatus() == 404) { + return false; + } + throw e; + } + } + void forgetDatabase(String name) { knownDatabases.remove(name); } diff --git a/streampipes-couchdb-client/src/test/java/org/apache/streampipes/couchdb/client/CouchDbClientIntegrationTest.java b/streampipes-couchdb-client/src/test/java/org/apache/streampipes/couchdb/client/CouchDbClientIntegrationTest.java index a5f8fe8e64..9a6bdc2b77 100644 --- a/streampipes-couchdb-client/src/test/java/org/apache/streampipes/couchdb/client/CouchDbClientIntegrationTest.java +++ b/streampipes-couchdb-client/src/test/java/org/apache/streampipes/couchdb/client/CouchDbClientIntegrationTest.java @@ -45,8 +45,13 @@ import org.testcontainers.containers.wait.strategy.Wait; import java.io.IOException; import java.io.InputStream; import java.net.ServerSocket; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; import java.nio.charset.StandardCharsets; import java.time.Duration; +import java.util.Base64; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -140,6 +145,35 @@ public class CouchDbClientIntegrationTest { assertEquals(2, db.raw().findAll().size()); } + @Test + public void usesExistingDatabaseWithoutTheRightToCreateDatabases() throws Exception { + String name = newName(); + server.database(name).create(); + openToAnonymousUsers(name); + + try (var anonymous = CouchServer.builder().url(server.config().baseUrl()).build()) { + var db = anonymous.database(name, DatabaseOptions.createIfMissing()); + db.ensureExists(); + assertEquals("a", db.raw().create(Map.of("_id", "a")).id()); + + var missing = anonymous.database(newName(), DatabaseOptions.createIfMissing()); + assertThrows(CouchDbAuthenticationException.class, missing::ensureExists); + } + } + + private static void openToAnonymousUsers(String database) throws Exception { + String auth = Base64.getEncoder().encodeToString((USER + ":" + PASSWORD).getBytes(StandardCharsets.UTF_8)); + try (var http = HttpClient.newHttpClient()) { + var uri = URI.create(server.config().baseUrl() + "/" + database + "/_security"); + var response = http.send(HttpRequest.newBuilder(uri) + .header("Authorization", "Basic " + auth) + .header("Content-Type", "application/json") + .PUT(HttpRequest.BodyPublishers.ofString("{\"members\":{\"names\":[],\"roles\":[]}}")) + .build(), HttpResponse.BodyHandlers.ofString()); + assertEquals(200, response.statusCode(), response.body()); + } + } + @Test public void writesAndReadsJacksonDocuments() { var docs = newDatabase().typed(JacksonDoc.class, JacksonCodec.defaults()); diff --git a/streampipes-couchdb-client/src/test/java/org/apache/streampipes/couchdb/client/TypedCollectionTest.java b/streampipes-couchdb-client/src/test/java/org/apache/streampipes/couchdb/client/TypedCollectionTest.java index b70ae9e942..de479adb50 100644 --- a/streampipes-couchdb-client/src/test/java/org/apache/streampipes/couchdb/client/TypedCollectionTest.java +++ b/streampipes-couchdb-client/src/test/java/org/apache/streampipes/couchdb/client/TypedCollectionTest.java @@ -151,26 +151,49 @@ public class TypedCollectionTest { @Test public void createsDatabaseOncePerServer() { var stub = StubTransport.create() - .respond(412, "{\"error\":\"file_exists\"}") + .respond(404, "") + .respond(201, "{\"ok\":true}") .respond(200, "{\"_id\":\"a\"}") .respond(200, "{\"_id\":\"b\"}"); var server = stub.server(); server.database("db", DatabaseOptions.createIfMissing()).raw().get("a"); server.database("db", DatabaseOptions.createIfMissing()).raw().get("b"); - assertEquals(List.of("PUT /db", "GET /db/a", "GET /db/b"), + assertEquals(List.of("HEAD /db", "PUT /db", "GET /db/a", "GET /db/b"), + stub.requests.stream().map(r -> r.method() + " " + r.path()).toList()); + } + + @Test + public void doesNotCreateAnExistingDatabase() { + var stub = StubTransport.create() + .respond(200, "") + .respond(200, "{\"_id\":\"a\"}"); + stub.server().database("db", DatabaseOptions.createIfMissing()).raw().get("a"); + assertEquals(List.of("HEAD /db", "GET /db/a"), + stub.requests.stream().map(r -> r.method() + " " + r.path()).toList()); + } + + @Test + public void acceptsDatabaseCreatedByAnotherClientAfterTheCheck() { + var stub = StubTransport.create() + .respond(404, "") + .respond(412, "{\"error\":\"file_exists\"}") + .respond(200, "{\"_id\":\"a\"}"); + stub.server().database("db", DatabaseOptions.createIfMissing()).raw().get("a"); + assertEquals(List.of("HEAD /db", "PUT /db", "GET /db/a"), stub.requests.stream().map(r -> r.method() + " " + r.path()).toList()); } @Test public void recreatesDeletedDatabaseAndRetries() { var stub = StubTransport.create() - .respond(201, "{\"ok\":true}") + .respond(200, "") .respond(404, "{\"error\":\"not_found\",\"reason\":\"Database does not exist.\"}") + .respond(404, "") .respond(201, "{\"ok\":true}") .respond(201, OK.formatted("a", "1-a")); var ref = stub.server().database("db", DatabaseOptions.createIfMissing()).raw().create(Map.of("_id", "a")); assertEquals("1-a", ref.rev()); - assertEquals(4, stub.requests.size()); + assertEquals(5, stub.requests.size()); } @Test diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/ImageStore.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/ImageStore.java index a7fbf9c595..27b4b090c4 100644 --- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/ImageStore.java +++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/ImageStore.java @@ -49,7 +49,7 @@ public class ImageStore { CouchServer couchServer) { this.couchServer = couchServer; var database = couchServer.database(DB_NAME, DatabaseOptions.createIfMissing()); - database.create(); + database.ensureExists(); this.attachments = database.attachments(); this.imageProperties = ImageStoreUtils.getImageProperties(measure); } diff --git a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/ImageStoreTest.java b/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/ImageStoreTest.java index 7c5f362142..603b273f4f 100644 --- a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/ImageStoreTest.java +++ b/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/ImageStoreTest.java @@ -51,13 +51,19 @@ class ImageStoreTest { private final CouchServer server = CouchServer.builder().transport(config -> transport).build(); @Test - void createsTheDatabaseWhenStartedAndAcceptsAnExistingOne() { - transport.databaseCreateStatus = 412; + void createsAMissingDatabaseWhenStarted() { + transport.databaseExists = false; new ImageStore(measure(), server); - assertEquals("PUT", transport.requests.get(0).method()); - assertEquals("/images", transport.requests.get(0).path()); + assertEquals(List.of("HEAD /images", "PUT /images"), transport.requestLines()); + } + + @Test + void doesNotCreateAnExistingDatabase() { + new ImageStore(measure(), server); + + assertEquals(List.of("HEAD /images"), transport.requestLines()); } @Test @@ -113,16 +119,14 @@ class ImageStoreTest { private static class RecordingTransport implements CouchTransport { private final List<CouchRequest> requests = new ArrayList<>(); - private int databaseCreateStatus = 201; + private boolean databaseExists = true; private boolean closed; @Override public CouchResponse send(CouchRequest request, Map<String, String> headers) { requests.add(request); - int status = "/images".equals(request.path()) ? databaseCreateStatus : 201; - String body = status == 201 - ? "{\"ok\":true,\"id\":\"id\",\"rev\":\"1-a\"}" - : "{\"error\":\"file_exists\",\"reason\":\"exists\"}"; + int status = "HEAD".equals(request.method()) ? (databaseExists ? 200 : 404) : 201; + String body = "{\"ok\":true,\"id\":\"id\",\"rev\":\"1-a\"}"; return new CouchResponse(status, Map.of(), new ByteArrayInputStream(body.getBytes(StandardCharsets.UTF_8))); } @@ -130,5 +134,9 @@ class ImageStoreTest { public void close() { closed = true; } + + private List<String> requestLines() { + return requests.stream().map(r -> r.method() + " " + r.path()).toList(); + } } } diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/couchdb/CouchDbSink.java b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/couchdb/CouchDbSink.java index 8d2471ce20..ae83be4578 100644 --- a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/couchdb/CouchDbSink.java +++ b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/couchdb/CouchDbSink.java @@ -97,7 +97,7 @@ public class CouchDbSink implements IStreamPipesDataSink { .connectTimeout(CONNECT_TIMEOUT) .build(); var database = couchServer.database(dbName, DatabaseOptions.createIfMissing()); - database.create(); + database.ensureExists(); // Gson keeps the stored format: fields without a value are left out. this.documents = database.typed(Object.class, GsonCodec.of(new Gson())); } diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/src/test/java/org/apache/streampipes/sinks/databases/jvm/couchdb/CouchDbSinkTest.java b/streampipes-extensions/streampipes-sinks-databases-jvm/src/test/java/org/apache/streampipes/sinks/databases/jvm/couchdb/CouchDbSinkTest.java index f14f9319d8..17e8ad56d7 100644 --- a/streampipes-extensions/streampipes-sinks-databases-jvm/src/test/java/org/apache/streampipes/sinks/databases/jvm/couchdb/CouchDbSinkTest.java +++ b/streampipes-extensions/streampipes-sinks-databases-jvm/src/test/java/org/apache/streampipes/sinks/databases/jvm/couchdb/CouchDbSinkTest.java @@ -48,7 +48,7 @@ class CouchDbSinkTest { private final List<Request> requests = new CopyOnWriteArrayList<>(); private final CouchDbSink sink = new CouchDbSink(); private HttpServer server; - private int databaseCreateStatus = 201; + private int databaseExistsStatus = 404; @BeforeEach void startServer() throws IOException { @@ -67,19 +67,18 @@ class CouchDbSinkTest { sink.onPipelineStarted(parameters(), null); sink.onPipelineStopped(); - assertEquals(1, requests.size()); - assertEquals("PUT /events", requests.get(0).line()); - assertNull(requests.get(0).authorization()); + assertEquals(List.of("HEAD /events", "PUT /events"), requests.stream().map(Request::line).toList()); + assertNull(requests.get(1).authorization()); } @Test - void acceptsAnExistingDatabase() { - databaseCreateStatus = 412; + void doesNotCreateAnExistingDatabase() { + databaseExistsStatus = 200; sink.onPipelineStarted(parameters(), null); sink.onPipelineStopped(); - assertEquals("PUT /events", requests.get(0).line()); + assertEquals(List.of("HEAD /events"), requests.stream().map(Request::line).toList()); } @Test @@ -116,7 +115,12 @@ class CouchDbSinkTest { String body = new String(exchange.getRequestBody().readAllBytes(), StandardCharsets.UTF_8); String line = exchange.getRequestMethod() + " " + exchange.getRequestURI().getRawPath(); requests.add(new Request(line, exchange.getRequestHeaders().getFirst("Authorization"), body)); - int status = "PUT /events".equals(line) ? databaseCreateStatus : 201; + if ("HEAD /events".equals(line)) { + exchange.sendResponseHeaders(databaseExistsStatus, -1); + exchange.close(); + return; + } + int status = 201; byte[] response = (status == 201 ? "{\"ok\":true,\"id\":\"id\",\"rev\":\"1-a\"}" : "{\"error\":\"file_exists\"}") .getBytes(StandardCharsets.UTF_8); exchange.sendResponseHeaders(status, response.length);
