This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 065646714889 CAMEL-26221: camel-ibm-watsonx-data - getAllColumns
parses its response with the managed com.ibm.cloud:sdk-core, and updateTable
takes the table patch as JSON text as well as a Map (#28975)
065646714889 is described below
commit 065646714889df4fad2995bf7845dceb32f1182e
Author: Andrea Cosentino <[email protected]>
AuthorDate: Mon Nov 2 23:21:39 2026 +0100
CAMEL-26221: camel-ibm-watsonx-data - getAllColumns parses its response
with the managed com.ibm.cloud:sdk-core, and updateTable takes the table patch
as JSON text as well as a Map (#28975)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/catalog/components/ibm-watsonx-data.json | 2 +-
.../catalog/docs/ibm-watsonx-data-component.adoc | 63 ++++-
.../camel-ibm/camel-ibm-watsonx-data/pom.xml | 6 +
.../ibm/watsonx/data/ibm-watsonx-data.json | 2 +-
.../src/main/docs/ibm-watsonx-data-component.adoc | 63 ++++-
.../ibm/watsonx/data/WatsonxDataEndpoint.java | 9 +-
.../ibm/watsonx/data/WatsonxDataProducer.java | 50 +++-
.../ibm/watsonx/data/WatsonxDataProducerTest.java | 255 +++++++++++++++++++++
8 files changed, 438 insertions(+), 12 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/ibm-watsonx-data.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/ibm-watsonx-data.json
index 17fe9eb88fa9..ecd044adede6 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/ibm-watsonx-data.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/ibm-watsonx-data.json
@@ -53,7 +53,7 @@
"CamelIBMWatsonxDataAuthInstanceId": { "index": 13, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "String", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "The auth instance ID for watsonx.data
API calls", "constantName":
"org.apache.camel.component.ibm.watsonx.data.WatsonxDataConstants#AUTH_INSTANCE_ID"
}
},
"bodyTypes": {
- "producerIn": { "carriers": [ "any" ], "byOption": [ { "option":
"operation", "value": "updateTable", "carriers": [ "map" ] } ], "description":
"Only updateTable reads the body: the patch of the table as a Map, sent as
JSON. JSON text, bytes or no body fail: unmarshal the JSON to a Map first. The
other operations take their input from the options and headers" },
+ "producerIn": { "carriers": [ "any" ], "byOption": [ { "option":
"operation", "value": "updateTable", "carriers": [ "map", "text", "bytes",
"stream", "file" ] } ], "description": "Only updateTable reads the body: the
patch of the table, as a Map or as a JSON object in text, bytes, a stream or a
file, sent as a JSON merge patch. No body, an empty patch or a body that is not
a JSON object fails. The other operations take their input from the options and
headers" },
"producerOut": { "carriers": [ "pojo" ], "byOption": [ { "option":
"operation", "value": "deleteCatalog", "carriers": [ "null" ] }, { "option":
"operation", "value": "deleteSchema", "carriers": [ "null" ] }, { "option":
"operation", "value": "deleteTable", "carriers": [ "null" ] } ], "description":
"The response of the IBM watsonx.data SDK, such as a CatalogCollection for
listCatalogs or a Table for getTable and updateTable; converted to text, it is
its JSON. The delete operations se [...]
},
"properties": {
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/ibm-watsonx-data-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/ibm-watsonx-data-component.adoc
index 4aa04ac90f44..00d3f65c8156 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/ibm-watsonx-data-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/ibm-watsonx-data-component.adoc
@@ -169,7 +169,7 @@ The component supports the following operations:
| `listTables` | List tables in a schema. Requires `catalogName`,
`schemaName`, and `engineId`.
| `getTable` | Get table details. Requires `catalogName`, `schemaName`,
`engineId`, and `CamelIBMWatsonxDataTableName` header.
| `deleteTable` | Delete a table. Requires `catalogName`, `schemaName`,
`engineId`, and `CamelIBMWatsonxDataTableName` header.
-| `updateTable` | Update table properties. Requires `catalogName`,
`schemaName`, `engineId`, `CamelIBMWatsonxDataTableName` header, and a `Map`
body with patch data.
+| `updateTable` | Update table properties. Requires `catalogName`,
`schemaName`, `engineId`, `CamelIBMWatsonxDataTableName` header, and the patch
of the table in the body: a `Map`, or a JSON object as text (a `String`,
`byte[]`, `InputStream` or file), which is sent as a JSON merge patch.
| `registerTable` | Register an external Iceberg table. Requires
`CamelIBMWatsonxDataCatalogId`, `CamelIBMWatsonxDataSchemaId`,
`CamelIBMWatsonxDataMetadataLocation`, and `CamelIBMWatsonxDataTableName`
headers.
| `getAllColumns` | Get all columns for a catalog. Requires `catalogName`.
Optionally filter by `schemaName` and `CamelIBMWatsonxDataTableName`.
|===
@@ -346,6 +346,67 @@ YAML::
----
====
+=== Rename a Table
+
+The patch of the table can be a JSON object as text, for example from a file
or an HTTP request:
+
+[tabs]
+====
+Java::
++
+[source,java]
+----
+from("direct:renameTable")
+ .setHeader("CamelIBMWatsonxDataTableName", constant("orders"))
+ .setBody(constant("{\"name\": \"orders_v2\"}"))
+
.to("ibm-watsonx-data:lakehouse?apiKey={{ibm.apiKey}}&serviceUrl={{ibm.watsonxdata.url}}&catalogName=my-catalog&schemaName=my-schema&engineId=my-presto-engine&operation=updateTable")
+ .log("Table updated: ${body}");
+----
+
+XML::
++
+[source,xml]
+----
+<route>
+ <from uri="direct:renameTable"/>
+ <setHeader name="CamelIBMWatsonxDataTableName">
+ <constant>orders</constant>
+ </setHeader>
+ <setBody>
+ <constant>{"name": "orders_v2"}</constant>
+ </setBody>
+ <to
uri="ibm-watsonx-data:lakehouse?apiKey={{ibm.apiKey}}&serviceUrl={{ibm.watsonxdata.url}}&catalogName=my-catalog&schemaName=my-schema&engineId=my-presto-engine&operation=updateTable"/>
+ <log message="Table updated: ${body}"/>
+</route>
+----
+
+YAML::
++
+[source,yaml]
+----
+- route:
+ from:
+ uri: direct:renameTable
+ steps:
+ - setHeader:
+ name: CamelIBMWatsonxDataTableName
+ constant: orders
+ - setBody:
+ constant: '{"name": "orders_v2"}'
+ - to:
+ uri: ibm-watsonx-data:lakehouse
+ parameters:
+ apiKey: "{{ibm.apiKey}}"
+ serviceUrl: "{{ibm.watsonxdata.url}}"
+ catalogName: my-catalog
+ schemaName: my-schema
+ engineId: my-presto-engine
+ operation: updateTable
+ - log:
+ message: "Table updated: ${body}"
+----
+====
+
=== Using Dynamic Operations via Headers
._Java-only: requires Java enum WatsonxDataOperations constant_
diff --git a/components/camel-ibm/camel-ibm-watsonx-data/pom.xml
b/components/camel-ibm/camel-ibm-watsonx-data/pom.xml
index 00f508856a27..b8e59a9d7170 100644
--- a/components/camel-ibm/camel-ibm-watsonx-data/pom.xml
+++ b/components/camel-ibm/camel-ibm-watsonx-data/pom.xml
@@ -44,6 +44,12 @@
<artifactId>watson-x-data</artifactId>
<version>${ibm-watsonx-data-sdk-version}</version>
</dependency>
+ <!-- the IBM SDK core at the version managed in camel-ibm-parent,
declared here so that applications
+ get it too, instead of the older one of the watsonx.data SDK that
fails with Gson 2.13 or later -->
+ <dependency>
+ <groupId>com.ibm.cloud</groupId>
+ <artifactId>sdk-core</artifactId>
+ </dependency>
<!-- for testing -->
<dependency>
diff --git
a/components/camel-ibm/camel-ibm-watsonx-data/src/generated/resources/META-INF/org/apache/camel/component/ibm/watsonx/data/ibm-watsonx-data.json
b/components/camel-ibm/camel-ibm-watsonx-data/src/generated/resources/META-INF/org/apache/camel/component/ibm/watsonx/data/ibm-watsonx-data.json
index 17fe9eb88fa9..ecd044adede6 100644
---
a/components/camel-ibm/camel-ibm-watsonx-data/src/generated/resources/META-INF/org/apache/camel/component/ibm/watsonx/data/ibm-watsonx-data.json
+++
b/components/camel-ibm/camel-ibm-watsonx-data/src/generated/resources/META-INF/org/apache/camel/component/ibm/watsonx/data/ibm-watsonx-data.json
@@ -53,7 +53,7 @@
"CamelIBMWatsonxDataAuthInstanceId": { "index": 13, "kind": "header",
"displayName": "", "group": "producer", "label": "", "required": false,
"javaType": "String", "deprecated": false, "deprecationNote": "", "autowired":
false, "secret": false, "description": "The auth instance ID for watsonx.data
API calls", "constantName":
"org.apache.camel.component.ibm.watsonx.data.WatsonxDataConstants#AUTH_INSTANCE_ID"
}
},
"bodyTypes": {
- "producerIn": { "carriers": [ "any" ], "byOption": [ { "option":
"operation", "value": "updateTable", "carriers": [ "map" ] } ], "description":
"Only updateTable reads the body: the patch of the table as a Map, sent as
JSON. JSON text, bytes or no body fail: unmarshal the JSON to a Map first. The
other operations take their input from the options and headers" },
+ "producerIn": { "carriers": [ "any" ], "byOption": [ { "option":
"operation", "value": "updateTable", "carriers": [ "map", "text", "bytes",
"stream", "file" ] } ], "description": "Only updateTable reads the body: the
patch of the table, as a Map or as a JSON object in text, bytes, a stream or a
file, sent as a JSON merge patch. No body, an empty patch or a body that is not
a JSON object fails. The other operations take their input from the options and
headers" },
"producerOut": { "carriers": [ "pojo" ], "byOption": [ { "option":
"operation", "value": "deleteCatalog", "carriers": [ "null" ] }, { "option":
"operation", "value": "deleteSchema", "carriers": [ "null" ] }, { "option":
"operation", "value": "deleteTable", "carriers": [ "null" ] } ], "description":
"The response of the IBM watsonx.data SDK, such as a CatalogCollection for
listCatalogs or a Table for getTable and updateTable; converted to text, it is
its JSON. The delete operations se [...]
},
"properties": {
diff --git
a/components/camel-ibm/camel-ibm-watsonx-data/src/main/docs/ibm-watsonx-data-component.adoc
b/components/camel-ibm/camel-ibm-watsonx-data/src/main/docs/ibm-watsonx-data-component.adoc
index 4aa04ac90f44..00d3f65c8156 100644
---
a/components/camel-ibm/camel-ibm-watsonx-data/src/main/docs/ibm-watsonx-data-component.adoc
+++
b/components/camel-ibm/camel-ibm-watsonx-data/src/main/docs/ibm-watsonx-data-component.adoc
@@ -169,7 +169,7 @@ The component supports the following operations:
| `listTables` | List tables in a schema. Requires `catalogName`,
`schemaName`, and `engineId`.
| `getTable` | Get table details. Requires `catalogName`, `schemaName`,
`engineId`, and `CamelIBMWatsonxDataTableName` header.
| `deleteTable` | Delete a table. Requires `catalogName`, `schemaName`,
`engineId`, and `CamelIBMWatsonxDataTableName` header.
-| `updateTable` | Update table properties. Requires `catalogName`,
`schemaName`, `engineId`, `CamelIBMWatsonxDataTableName` header, and a `Map`
body with patch data.
+| `updateTable` | Update table properties. Requires `catalogName`,
`schemaName`, `engineId`, `CamelIBMWatsonxDataTableName` header, and the patch
of the table in the body: a `Map`, or a JSON object as text (a `String`,
`byte[]`, `InputStream` or file), which is sent as a JSON merge patch.
| `registerTable` | Register an external Iceberg table. Requires
`CamelIBMWatsonxDataCatalogId`, `CamelIBMWatsonxDataSchemaId`,
`CamelIBMWatsonxDataMetadataLocation`, and `CamelIBMWatsonxDataTableName`
headers.
| `getAllColumns` | Get all columns for a catalog. Requires `catalogName`.
Optionally filter by `schemaName` and `CamelIBMWatsonxDataTableName`.
|===
@@ -346,6 +346,67 @@ YAML::
----
====
+=== Rename a Table
+
+The patch of the table can be a JSON object as text, for example from a file
or an HTTP request:
+
+[tabs]
+====
+Java::
++
+[source,java]
+----
+from("direct:renameTable")
+ .setHeader("CamelIBMWatsonxDataTableName", constant("orders"))
+ .setBody(constant("{\"name\": \"orders_v2\"}"))
+
.to("ibm-watsonx-data:lakehouse?apiKey={{ibm.apiKey}}&serviceUrl={{ibm.watsonxdata.url}}&catalogName=my-catalog&schemaName=my-schema&engineId=my-presto-engine&operation=updateTable")
+ .log("Table updated: ${body}");
+----
+
+XML::
++
+[source,xml]
+----
+<route>
+ <from uri="direct:renameTable"/>
+ <setHeader name="CamelIBMWatsonxDataTableName">
+ <constant>orders</constant>
+ </setHeader>
+ <setBody>
+ <constant>{"name": "orders_v2"}</constant>
+ </setBody>
+ <to
uri="ibm-watsonx-data:lakehouse?apiKey={{ibm.apiKey}}&serviceUrl={{ibm.watsonxdata.url}}&catalogName=my-catalog&schemaName=my-schema&engineId=my-presto-engine&operation=updateTable"/>
+ <log message="Table updated: ${body}"/>
+</route>
+----
+
+YAML::
++
+[source,yaml]
+----
+- route:
+ from:
+ uri: direct:renameTable
+ steps:
+ - setHeader:
+ name: CamelIBMWatsonxDataTableName
+ constant: orders
+ - setBody:
+ constant: '{"name": "orders_v2"}'
+ - to:
+ uri: ibm-watsonx-data:lakehouse
+ parameters:
+ apiKey: "{{ibm.apiKey}}"
+ serviceUrl: "{{ibm.watsonxdata.url}}"
+ catalogName: my-catalog
+ schemaName: my-schema
+ engineId: my-presto-engine
+ operation: updateTable
+ - log:
+ message: "Table updated: ${body}"
+----
+====
+
=== Using Dynamic Operations via Headers
._Java-only: requires Java enum WatsonxDataOperations constant_
diff --git
a/components/camel-ibm/camel-ibm-watsonx-data/src/main/java/org/apache/camel/component/ibm/watsonx/data/WatsonxDataEndpoint.java
b/components/camel-ibm/camel-ibm-watsonx-data/src/main/java/org/apache/camel/component/ibm/watsonx/data/WatsonxDataEndpoint.java
index 1b29bd6872ba..dd0e2176cb95 100644
---
a/components/camel-ibm/camel-ibm-watsonx-data/src/main/java/org/apache/camel/component/ibm/watsonx/data/WatsonxDataEndpoint.java
+++
b/components/camel-ibm/camel-ibm-watsonx-data/src/main/java/org/apache/camel/component/ibm/watsonx/data/WatsonxDataEndpoint.java
@@ -46,9 +46,12 @@ import org.apache.camel.util.ObjectHelper;
headersClass = WatsonxDataConstants.class)
@BodyTypes(producerIn = @BodyType(value = Carrier.ANY,
byOption = @BodyType.When(option =
"operation", value = "updateTable",
- carriers =
Carrier.MAP),
- description = "Only updateTable reads the
body: the patch of the table as a Map, sent as JSON."
- + " JSON text, bytes or no
body fail: unmarshal the JSON to a Map first. The other"
+ carriers = {
+
Carrier.MAP, Carrier.TEXT, Carrier.BYTES, Carrier.STREAM,
+
Carrier.FILE }),
+ description = "Only updateTable reads the
body: the patch of the table, as a Map or as a JSON"
+ + " object in text, bytes, a
stream or a file, sent as a JSON merge patch. No"
+ + " body, an empty patch or a
body that is not a JSON object fails. The other"
+ " operations take their
input from the options and headers"),
producerOut = @BodyType(value = Carrier.POJO,
byOption = {
diff --git
a/components/camel-ibm/camel-ibm-watsonx-data/src/main/java/org/apache/camel/component/ibm/watsonx/data/WatsonxDataProducer.java
b/components/camel-ibm/camel-ibm-watsonx-data/src/main/java/org/apache/camel/component/ibm/watsonx/data/WatsonxDataProducer.java
index 5ff9ad02d1a9..baa379c9d94f 100644
---
a/components/camel-ibm/camel-ibm-watsonx-data/src/main/java/org/apache/camel/component/ibm/watsonx/data/WatsonxDataProducer.java
+++
b/components/camel-ibm/camel-ibm-watsonx-data/src/main/java/org/apache/camel/component/ibm/watsonx/data/WatsonxDataProducer.java
@@ -18,7 +18,12 @@ package org.apache.camel.component.ibm.watsonx.data;
import java.util.Map;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonParseException;
+import com.google.gson.JsonParser;
+import com.google.gson.reflect.TypeToken;
import com.ibm.cloud.sdk.core.http.Response;
+import com.ibm.cloud.sdk.core.util.GsonSingleton;
import com.ibm.cloud.watsonxdata.watsonx_data.v3.WatsonxData;
import com.ibm.cloud.watsonxdata.watsonx_data.v3.model.Catalog;
import com.ibm.cloud.watsonxdata.watsonx_data.v3.model.CatalogCollection;
@@ -64,6 +69,14 @@ public class WatsonxDataProducer extends DefaultProducer {
private static final Logger LOG =
LoggerFactory.getLogger(WatsonxDataProducer.class);
+ // the bodies updateTable takes, for its error messages
+ private static final String TABLE_PATCH_BODIES
+ = "a java.util.Map, or a JSON object as text (a String, byte[],
InputStream or file)";
+ private static final String TABLE_PATCH_NOT_JSON_OBJECT
+ = "The body of the updateTable operation is not a JSON object: it
must be " + TABLE_PATCH_BODIES;
+ private static final TypeToken<Map<String, Object>> TABLE_PATCH_TYPE = new
TypeToken<>() {
+ };
+
public WatsonxDataProducer(WatsonxDataEndpoint endpoint) {
super(endpoint);
}
@@ -261,7 +274,6 @@ public class WatsonxDataProducer extends DefaultProducer {
exchange.getMessage().setBody(null);
}
- @SuppressWarnings("unchecked")
private void handleUpdateTable(Exchange exchange, WatsonxData service,
WatsonxDataConfiguration config) {
String catalogName = resolveHeader(exchange,
WatsonxDataConstants.CATALOG_NAME, config.getCatalogName());
String schemaName = resolveHeader(exchange,
WatsonxDataConstants.SCHEMA_NAME, config.getSchemaName());
@@ -272,10 +284,7 @@ public class WatsonxDataProducer extends DefaultProducer {
requireNonEmpty(tableName, "Table name");
requireNonEmpty(engineId, "Engine ID");
- Map<String, Object> body = exchange.getIn().getBody(Map.class);
- if (body == null || body.isEmpty()) {
- throw new IllegalArgumentException("Request body with table patch
data is required for updateTable operation");
- }
+ Map<String, Object> body = getTablePatch(exchange);
UpdateTableOptions.Builder builder
= new UpdateTableOptions.Builder(catalogName, schemaName,
tableName, engineId, body);
@@ -285,6 +294,37 @@ public class WatsonxDataProducer extends DefaultProducer {
exchange.getMessage().setBody(response.getResult());
}
+ /**
+ * The patch of the table for updateTable: a Map, or a JSON object as
text, such as a String from HTTP or a file.
+ * The JSON is parsed with the Gson of the IBM SDK, which writes the patch
back to JSON for the request, so its
+ * numbers and nulls are sent as they are written.
+ */
+ private static Map<String, Object> getTablePatch(Exchange exchange) {
+ @SuppressWarnings("unchecked")
+ Map<String, Object> patch = exchange.getIn().getBody(Map.class);
+ if (patch == null && exchange.getIn().getBody() != null) {
+ String json = exchange.getIn().getBody(String.class);
+ if (json != null) {
+ JsonElement element;
+ try {
+ element = JsonParser.parseString(json);
+ } catch (JsonParseException e) {
+ throw new
IllegalArgumentException(TABLE_PATCH_NOT_JSON_OBJECT, e);
+ }
+ if (element.isJsonObject()) {
+ patch =
GsonSingleton.getGsonWithoutPrettyPrinting().fromJson(element,
TABLE_PATCH_TYPE);
+ } else if (!element.isJsonNull()) {
+ throw new
IllegalArgumentException(TABLE_PATCH_NOT_JSON_OBJECT);
+ }
+ }
+ }
+ if (patch == null || patch.isEmpty()) {
+ throw new IllegalArgumentException(
+ "Request body with table patch data is required for
updateTable operation: " + TABLE_PATCH_BODIES);
+ }
+ return patch;
+ }
+
private void handleRegisterTable(Exchange exchange, WatsonxData service,
WatsonxDataConfiguration config) {
String catalogId =
exchange.getIn().getHeader(WatsonxDataConstants.CATALOG_ID, String.class);
String schemaId =
exchange.getIn().getHeader(WatsonxDataConstants.SCHEMA_ID, String.class);
diff --git
a/components/camel-ibm/camel-ibm-watsonx-data/src/test/java/org/apache/camel/component/ibm/watsonx/data/WatsonxDataProducerTest.java
b/components/camel-ibm/camel-ibm-watsonx-data/src/test/java/org/apache/camel/component/ibm/watsonx/data/WatsonxDataProducerTest.java
new file mode 100644
index 000000000000..a58459461011
--- /dev/null
+++
b/components/camel-ibm/camel-ibm-watsonx-data/src/test/java/org/apache/camel/component/ibm/watsonx/data/WatsonxDataProducerTest.java
@@ -0,0 +1,255 @@
+/*
+ * 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.
+ */
+package org.apache.camel.component.ibm.watsonx.data;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.net.URLDecoder;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Instant;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Stream;
+
+import com.google.gson.JsonParser;
+import com.ibm.cloud.sdk.core.security.IamAuthenticator;
+import com.ibm.cloud.watsonxdata.watsonx_data.v3.model.ColumnsCollection;
+import com.ibm.cloud.watsonxdata.watsonx_data.v3.model.Table;
+import com.ibm.cloud.watsonxdata.watsonx_data.v3.model.TableColumDetail;
+import
com.ibm.cloud.watsonxdata.watsonx_data.v3.model.TableColumDetailColumnsItems;
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+import org.apache.camel.CamelExecutionException;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Named;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The getAllColumns and updateTable operations, with a JDK HttpServer as the
watsonx.data service and as the IAM token
+ * service.
+ */
+class WatsonxDataProducerTest extends CamelTestSupport {
+
+ private static final String TABLE_PATH =
"/catalogs/iceberg_data/schemas/sales/tables/orders";
+ private static final String TABLE_PATCH = "{\"name\":\"orders_v2\"}";
+ private static final String ACCEPTED_BODIES
+ = "a java.util.Map, or a JSON object as text (a String, byte[],
InputStream or file)";
+
+ @TempDir
+ Path dir;
+
+ private HttpServer server;
+ private final AtomicInteger patchRequests = new AtomicInteger();
+ private volatile String columnsRequestUri;
+ private volatile String patchRequestUri;
+ private volatile String patchContentType;
+ private volatile String patchBody;
+
+ @Override
+ protected void setupResources() throws Exception {
+ server = HttpServer.create(new InetSocketAddress("localhost", 0), 0);
+ // the IAM token service, which the IamAuthenticator of the endpoint
asks for a token for the API key
+ server.createContext("/identity/token", exchange -> {
+ exchange.getRequestBody().readAllBytes();
+ respond(exchange,
"{\"access_token\":\"test-access-token\",\"refresh_token\":\"test-refresh-token\","
+ +
"\"token_type\":\"Bearer\",\"expires_in\":3600,\"expiration\":"
+ + (Instant.now().getEpochSecond() + 3600) + "}");
+ });
+ server.createContext("/columns", exchange -> {
+ columnsRequestUri = exchange.getRequestURI().toString();
+ respond(exchange,
"{\"columns\":[{\"catalog\":\"iceberg_data\",\"schema\":\"sales\",\"table\":\"orders\","
+ +
"\"columns\":[{\"column\":\"id\",\"type\":\"bigint\",\"nullable\":false,\"index\":1,"
+ + "\"comment\":\"the order
id\"}]}],\"count\":1}");
+ });
+ server.createContext(TABLE_PATH, exchange -> {
+ patchRequests.incrementAndGet();
+ patchRequestUri = exchange.getRequestMethod() + " " +
exchange.getRequestURI();
+ patchContentType =
exchange.getRequestHeaders().getFirst("Content-Type");
+ patchBody = new String(exchange.getRequestBody().readAllBytes(),
StandardCharsets.UTF_8);
+ respond(exchange,
"{\"name\":\"orders_v2\",\"columns\":[{\"name\":\"id\",\"type\":\"bigint\"}]}");
+ });
+ server.start();
+ }
+
+ @Override
+ protected void cleanupResources() throws Exception {
+ if (server != null) {
+ server.stop(0);
+ }
+ }
+
+ @BeforeEach
+ void useStubTokenService() {
+ // the endpoint authenticates with an IAM API key: take the token from
the stub instead of IBM Cloud
+ WatsonxDataEndpoint endpoint = context.getEndpoint(watsonxDataUri(),
WatsonxDataEndpoint.class);
+ ((IamAuthenticator)
endpoint.getWatsonxDataService().getAuthenticator()).setURL(serverUrl());
+ }
+
+ @Test
+ void getAllColumnsParsesColumnsWithDynamicProperties() {
+ Object body = template.requestBodyAndHeader("direct:start", null,
WatsonxDataConstants.OPERATION,
+ WatsonxDataOperations.getAllColumns);
+
+ ColumnsCollection columns = assertInstanceOf(ColumnsCollection.class,
body);
+ assertEquals(1L, columns.getCount());
+ TableColumDetail table = columns.getColumns().get(0);
+ assertEquals("orders", table.getTable());
+ TableColumDetailColumnsItems column = table.getColumns().get(0);
+ assertEquals("id", column.getColumn());
+ assertEquals("bigint", column.getType());
+ assertFalse(column.isNullable());
+ // a property the model does not declare is kept as a dynamic property
of the column
+ assertEquals("the order id", column.get("comment"));
+
+ assertTrue(columnsRequestUri.startsWith("/columns?"),
columnsRequestUri);
+ assertEquals(Map.of("catalog_name", "iceberg_data", "schema_name",
"sales"), queryParameters(columnsRequestUri));
+ }
+
+ @Test
+ void updateTableSendsMapBody() {
+ Table table = updateTable(Map.of("name", "orders_v2"));
+
+ assertEquals("orders_v2", table.getName());
+ assertPatchSent();
+ }
+
+ static Stream<Named<Object>> jsonTextBodies() {
+ return Stream.of(
+ Named.of("String", TABLE_PATCH),
+ Named.of("byte[]",
TABLE_PATCH.getBytes(StandardCharsets.UTF_8)),
+ Named.of("InputStream", new
ByteArrayInputStream(TABLE_PATCH.getBytes(StandardCharsets.UTF_8))));
+ }
+
+ @ParameterizedTest
+ @MethodSource("jsonTextBodies")
+ void updateTableSendsJsonTextBody(Object body) {
+ Table table = updateTable(body);
+
+ assertEquals("orders_v2", table.getName());
+ assertPatchSent();
+ }
+
+ @Test
+ void updateTableSendsJsonFileBody() throws IOException {
+ Path file = Files.writeString(dir.resolve("patch.json"), TABLE_PATCH);
+
+ Table table = updateTable(file.toFile());
+
+ assertEquals("orders_v2", table.getName());
+ assertPatchSent();
+ }
+
+ @Test
+ void updateTableWithoutBodyFails() {
+ CamelExecutionException e =
assertThrows(CamelExecutionException.class, () -> updateTable(null));
+
+ IllegalArgumentException cause =
assertInstanceOf(IllegalArgumentException.class, e.getCause());
+ assertEquals("Request body with table patch data is required for
updateTable operation: " + ACCEPTED_BODIES,
+ cause.getMessage());
+ assertEquals(0, patchRequests.get());
+ }
+
+ static Stream<Named<Object>> bodiesThatAreNotJsonObjects() {
+ return Stream.of(
+ Named.of("JSON array", "[\"orders_v2\"]"),
+ Named.of("text that is not JSON", "name=orders_v2"),
+ Named.of("bytes that are not UTF-8", new byte[] { (byte) 0xff,
(byte) 0xfe }),
+ Named.of("POJO", new TableRename("orders_v2")));
+ }
+
+ @ParameterizedTest
+ @MethodSource("bodiesThatAreNotJsonObjects")
+ void updateTableWithBodyThatIsNotJsonObjectFails(Object body) {
+ CamelExecutionException e =
assertThrows(CamelExecutionException.class, () -> updateTable(body));
+
+ IllegalArgumentException cause =
assertInstanceOf(IllegalArgumentException.class, e.getCause());
+ assertEquals("The body of the updateTable operation is not a JSON
object: it must be " + ACCEPTED_BODIES,
+ cause.getMessage());
+ assertEquals(0, patchRequests.get());
+ }
+
+ private Table updateTable(Object body) {
+ Map<String, Object> headers = new HashMap<>();
+ headers.put(WatsonxDataConstants.OPERATION,
WatsonxDataOperations.updateTable);
+ headers.put(WatsonxDataConstants.TABLE_NAME, "orders");
+ return template.requestBodyAndHeaders("direct:start", body, headers,
Table.class);
+ }
+
+ private void assertPatchSent() {
+ assertEquals("PATCH " + TABLE_PATH + "?engine_id=presto-01",
patchRequestUri);
+
assertTrue(patchContentType.startsWith("application/merge-patch+json"),
patchContentType);
+ assertEquals(JsonParser.parseString(TABLE_PATCH),
JsonParser.parseString(patchBody));
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start").to(watsonxDataUri());
+ }
+ };
+ }
+
+ private String serverUrl() {
+ return "http://localhost:"; + server.getAddress().getPort();
+ }
+
+ private String watsonxDataUri() {
+ return
"ibm-watsonx-data:test?apiKey=test-api-key&catalogName=iceberg_data&schemaName=sales&engineId=presto-01"
+ + "&serviceUrl=" + serverUrl();
+ }
+
+ private static Map<String, String> queryParameters(String uri) {
+ Map<String, String> answer = new HashMap<>();
+ for (String pair : uri.substring(uri.indexOf('?') + 1).split("&")) {
+ String[] nameAndValue = pair.split("=", 2);
+ answer.put(URLDecoder.decode(nameAndValue[0],
StandardCharsets.UTF_8),
+ URLDecoder.decode(nameAndValue[1],
StandardCharsets.UTF_8));
+ }
+ return answer;
+ }
+
+ // a POJO the type converters cannot turn into a Map: its text is not JSON
+ record TableRename(String name) {
+ }
+
+ private static void respond(HttpExchange exchange, String json) throws
IOException {
+ byte[] body = json.getBytes(StandardCharsets.UTF_8);
+ exchange.getResponseHeaders().set("Content-Type", "application/json");
+ exchange.sendResponseHeaders(200, body.length);
+ try (OutputStream out = exchange.getResponseBody()) {
+ out.write(body);
+ }
+ }
+}
