You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ya...@apache.org on 2023/10/14 09:16:58 UTC
[kafka] branch trunk updated: KAFKA-15387: Deprecate Connect's redundant task configurations endpoint (#14361)
This is an automated email from the ASF dual-hosted git repository.
yashmayya pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 1c8bb61a43d KAFKA-15387: Deprecate Connect's redundant task configurations endpoint (#14361)
1c8bb61a43d is described below
commit 1c8bb61a43d3ad1fd7a10eb3947342ceba783c4e
Author: Yash Mayya <ya...@gmail.com>
AuthorDate: Sat Oct 14 10:16:50 2023 +0100
KAFKA-15387: Deprecate Connect's redundant task configurations endpoint (#14361)
Reviewers: Mickael Maison <mi...@gmail.com>, Sagar Rao <sa...@gmail.com>
---
.../connect/runtime/rest/entities/TaskInfo.java | 4 +-
.../runtime/rest/resources/ConnectorsResource.java | 4 +-
.../integration/ConnectWorkerIntegrationTest.java | 45 ++++++++++++++++++++--
.../util/clusters/EmbeddedConnectCluster.java | 11 +++---
docs/connect.html | 3 +-
5 files changed, 55 insertions(+), 12 deletions(-)
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/entities/TaskInfo.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/entities/TaskInfo.java
index 8e6f3d7baac..635b8be0704 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/entities/TaskInfo.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/entities/TaskInfo.java
@@ -16,6 +16,7 @@
*/
package org.apache.kafka.connect.runtime.rest.entities;
+import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import org.apache.kafka.connect.util.ConnectorTaskId;
@@ -26,7 +27,8 @@ public class TaskInfo {
private final ConnectorTaskId id;
private final Map<String, String> config;
- public TaskInfo(ConnectorTaskId id, Map<String, String> config) {
+ @JsonCreator
+ public TaskInfo(@JsonProperty("id") ConnectorTaskId id, @JsonProperty("config") Map<String, String> config) {
this.id = id;
this.config = config;
}
diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorsResource.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorsResource.java
index 9f1cabcfa14..4878b8df9e1 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorsResource.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorsResource.java
@@ -173,9 +173,11 @@ public class ConnectorsResource implements ConnectResource {
@GET
@Path("/{connector}/tasks-config")
- @Operation(summary = "Get the configuration of all tasks for the specified connector")
+ @Operation(deprecated = true, summary = "Get the configuration of all tasks for the specified connector")
public Map<ConnectorTaskId, Map<String, String>> getTasksConfig(
final @PathParam("connector") String connector) throws Throwable {
+ log.warn("The 'GET /connectors/{connector}/tasks-config' endpoint is deprecated and will be removed in the next major release. "
+ + "Please use the 'GET /connectors/{connector}/tasks' endpoint instead.");
FutureCallback<Map<ConnectorTaskId, Map<String, String>>> cb = new FutureCallback<>();
herder.tasksConfig(connector, cb);
return requestHandler.completeRequest(cb);
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectWorkerIntegrationTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectWorkerIntegrationTest.java
index 4c393d95ad3..42c3831faf4 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectWorkerIntegrationTest.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectWorkerIntegrationTest.java
@@ -16,7 +16,9 @@
*/
package org.apache.kafka.connect.integration;
+import org.apache.kafka.common.utils.LogCaptureAppender;
import org.apache.kafka.connect.runtime.distributed.DistributedConfig;
+import org.apache.kafka.connect.runtime.rest.resources.ConnectorsResource;
import org.apache.kafka.connect.storage.StringConverter;
import org.apache.kafka.connect.util.clusters.EmbeddedConnectCluster;
import org.apache.kafka.connect.util.clusters.WorkerHandle;
@@ -29,9 +31,11 @@ import org.junit.experimental.categories.Category;
import org.junit.rules.TestRule;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.slf4j.event.Level;
import java.util.Collections;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Properties;
@@ -51,6 +55,8 @@ import static org.apache.kafka.connect.runtime.TopicCreationConfig.REPLICATION_F
import static org.apache.kafka.connect.runtime.WorkerConfig.CONNECTOR_CLIENT_POLICY_CLASS_CONFIG;
import static org.apache.kafka.connect.runtime.WorkerConfig.OFFSET_COMMIT_INTERVAL_MS_CONFIG;
import static org.apache.kafka.connect.util.clusters.EmbeddedConnectClusterAssertions.CONNECTOR_SETUP_DURATION_MS;
+import static org.hamcrest.CoreMatchers.containsString;
+import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
@@ -382,7 +388,7 @@ public class ConnectWorkerIntegrationTest {
);
// If the connector is truly stopped, we should also see an empty set of tasks and task configs
assertEquals(Collections.emptyList(), connect.connectorInfo(CONNECTOR_NAME).tasks());
- assertEquals(Collections.emptyMap(), connect.taskConfigs(CONNECTOR_NAME));
+ assertEquals(Collections.emptyList(), connect.taskConfigs(CONNECTOR_NAME));
// Transition to RUNNING
connect.resumeConnector(CONNECTOR_NAME);
@@ -411,7 +417,7 @@ public class ConnectWorkerIntegrationTest {
"Connector did not stop in time"
);
assertEquals(Collections.emptyList(), connect.connectorInfo(CONNECTOR_NAME).tasks());
- assertEquals(Collections.emptyMap(), connect.taskConfigs(CONNECTOR_NAME));
+ assertEquals(Collections.emptyList(), connect.taskConfigs(CONNECTOR_NAME));
// Transition to PAUSED
connect.pauseConnector(CONNECTOR_NAME);
@@ -471,7 +477,7 @@ public class ConnectWorkerIntegrationTest {
);
// If the connector is truly stopped, we should also see an empty set of tasks and task configs
assertEquals(Collections.emptyList(), connect.connectorInfo(CONNECTOR_NAME).tasks());
- assertEquals(Collections.emptyMap(), connect.taskConfigs(CONNECTOR_NAME));
+ assertEquals(Collections.emptyList(), connect.taskConfigs(CONNECTOR_NAME));
// Can resume a connector after its Connector has failed before shutdown after receiving a stop request
props.remove("connector.start.inject.error");
@@ -493,7 +499,7 @@ public class ConnectWorkerIntegrationTest {
"Connector did not stop in time"
);
assertEquals(Collections.emptyList(), connect.connectorInfo(CONNECTOR_NAME).tasks());
- assertEquals(Collections.emptyMap(), connect.taskConfigs(CONNECTOR_NAME));
+ assertEquals(Collections.emptyList(), connect.taskConfigs(CONNECTOR_NAME));
// Can resume a connector after its Connector has failed during shutdown after receiving a stop request
connect.resumeConnector(CONNECTOR_NAME);
@@ -511,6 +517,37 @@ public class ConnectWorkerIntegrationTest {
);
}
+ /**
+ * The <strong><em>GET /connectors/{connector}/tasks-config</em></strong> endpoint was deprecated in
+ * <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-970%3A+Deprecate+and+remove+Connect%27s+redundant+task+configurations+endpoint">KIP-970</a>
+ * and is slated for removal in the next major release. This test verifies that the deprecation warning log is emitted on trying to use the
+ * deprecated endpoint.
+ */
+ @Test
+ public void testTasksConfigDeprecation() throws Exception {
+ connect = connectBuilder.build();
+ // start the clusters
+ connect.start();
+
+ connect.assertions().assertAtLeastNumWorkersAreUp(NUM_WORKERS,
+ "Initial group of workers did not start in time.");
+
+ connect.configureConnector(CONNECTOR_NAME, defaultSourceConnectorProps(TOPIC_NAME));
+ connect.assertions().assertConnectorAndExactlyNumTasksAreRunning(
+ CONNECTOR_NAME,
+ NUM_TASKS,
+ "Connector tasks did not start in time"
+ );
+
+ try (LogCaptureAppender logCaptureAppender = LogCaptureAppender.createAndRegister(ConnectorsResource.class)) {
+ connect.requestGet(connect.endpointForResource("connectors/" + CONNECTOR_NAME + "/tasks-config"));
+ List<LogCaptureAppender.Event> logEvents = logCaptureAppender.getEvents();
+ assertEquals(1, logEvents.size());
+ assertEquals(Level.WARN.toString(), logEvents.get(0).getLevel());
+ assertThat(logEvents.get(0).getMessage(), containsString("deprecated"));
+ }
+ }
+
private Map<String, String> defaultSourceConnectorProps(String topic) {
// setup up props for the source connector
Map<String, String> props = new HashMap<>();
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedConnectCluster.java b/connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedConnectCluster.java
index 20dce332fcf..9e3edcd8f24 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedConnectCluster.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/util/clusters/EmbeddedConnectCluster.java
@@ -31,6 +31,7 @@ import org.apache.kafka.connect.runtime.rest.entities.ConnectorOffset;
import org.apache.kafka.connect.runtime.rest.entities.ConnectorOffsets;
import org.apache.kafka.connect.runtime.rest.entities.ConnectorStateInfo;
import org.apache.kafka.connect.runtime.rest.entities.ServerInfo;
+import org.apache.kafka.connect.runtime.rest.entities.TaskInfo;
import org.apache.kafka.connect.runtime.rest.errors.ConnectRestException;
import org.apache.kafka.connect.util.SinkUtils;
import org.eclipse.jetty.client.HttpClient;
@@ -611,19 +612,19 @@ public class EmbeddedConnectCluster {
/**
* Get the task configs of a connector running in this cluster.
-
+ *
* @param connectorName name of the connector
- * @return a map from task ID (connector name + "-" + task number) to task config
+ * @return a list of task configurations for the connector
*/
- public Map<String, Map<String, String>> taskConfigs(String connectorName) {
+ public List<TaskInfo> taskConfigs(String connectorName) {
ObjectMapper mapper = new ObjectMapper();
- String url = endpointForResource(String.format("connectors/%s/tasks-config", connectorName));
+ String url = endpointForResource(String.format("connectors/%s/tasks", connectorName));
Response response = requestGet(url);
try {
if (response.getStatus() < Response.Status.BAD_REQUEST.getStatusCode()) {
// We use String instead of ConnectorTaskId as the key here since the latter can't be automatically
// deserialized by Jackson when used as a JSON object key (i.e., when it's serialized as a JSON string)
- return mapper.readValue(responseToString(response), new TypeReference<Map<String, Map<String, String>>>() { });
+ return mapper.readValue(responseToString(response), new TypeReference<List<TaskInfo>>() { });
}
} catch (IOException e) {
log.error("Could not read task configs from response: {}",
diff --git a/docs/connect.html b/docs/connect.html
index 2deb8901888..16f268e78b0 100644
--- a/docs/connect.html
+++ b/docs/connect.html
@@ -298,7 +298,8 @@ listeners=http://localhost:8080,https://localhost:8443</pre>
<li><code>GET /connectors/{name}/config</code> - get the configuration parameters for a specific connector</li>
<li><code>PUT /connectors/{name}/config</code> - update the configuration parameters for a specific connector</li>
<li><code>GET /connectors/{name}/status</code> - get current status of the connector, including if it is running, failed, paused, etc., which worker it is assigned to, error information if it has failed, and the state of all its tasks</li>
- <li><code>GET /connectors/{name}/tasks</code> - get a list of tasks currently running for a connector</li>
+ <li><code>GET /connectors/{name}/tasks</code> - get a list of tasks currently running for a connector along with their configurations</li>
+ <li><code>GET /connectors/{name}/tasks-config</code> - get the configuration of all tasks for a specific connector. This endpoint is deprecated and will be removed in the next major release. Please use the <code>GET /connectors/{name}/tasks</code> endpoint instead. Note that the response structures of the two endpoints differ slightly, please refer to the <a href="/{{version}}/generated/connect_rest.yaml">OpenAPI documentation</a> for more details</li>
<li><code>GET /connectors/{name}/tasks/{taskid}/status</code> - get current status of the task, including if it is running, failed, paused, etc., which worker it is assigned to, and error information if it has failed</li>
<li><code>PUT /connectors/{name}/pause</code> - pause the connector and its tasks, which stops message processing until the connector is resumed. Any resources claimed by its tasks are left allocated, which allows the connector to begin processing data quickly once it is resumed.</li>
<li id="connect_stopconnector"><code>PUT /connectors/{name}/stop</code> - stop the connector and shut down its tasks, deallocating any resources claimed by its tasks. This is more efficient from a resource usage standpoint than pausing the connector, but can cause it to take longer to begin processing data once resumed. Note that the offsets for a connector can be only modified via the offsets management endpoints if it is in the stopped state</li>