### MaxCompute Kafka Connector Configuration Example

Source: https://github.com/aliyun/maxcompute-kafka-connector/blob/master/README.md

This is a sample configuration file for the MaxCompute Kafka Connector, demonstrating how to set up the connector to ingest data from Kafka topics into MaxCompute tables. It includes parameters for general connector settings, topic and MaxCompute connection details, authentication credentials, data format, partitioning, and error handling.

```json
{
    "name": "your_name",
    "config": {
	"connector.class": "com.aliyun.odps.kafka.connect.MaxComputeSinkConnector",
	"tasks.max": "3",
	"topics": "your_topic",
	"endpoint": "endpoint",
	"tunnel_endpoint": "your_tunnel endpoint",
	"project": "project",
	"table": "your_table",
	"account_type": "account type (STS or ALIYUN)",
	"access_id": "access id",
	"access_key": "access key",
	"account_id": "account id for sts",
	"sts.endpoint": "sts endpoint",
	"region_id": "region id for sts",
	"role_name": "role name for sts",
	"client_timeout_ms": "STS Token valid period (ms)",
	"format": "TEXT",
	"mode": "KEY",
	"partition_window_type": "MINUTE",
	"use_new_partition_format":true,
	"use_streaming": false,
	"buffer_size_kb": 65536,
	"sink_pool_size":"16",
	"record_batch_size":"8000",
	"runtime.error.topic.name":"kafka topic when runtime errors happens",
	"runtime.error.topic.bootstrap.servers":"kafka bootstrap servers of error topic queue",
	"skip_error":"false"
    }
}

```

--------------------------------

### Java MaxCompute Sink Writer Example

Source: https://context7.com/aliyun/maxcompute-kafka-connector/llms.txt

Demonstrates how to use the MaxComputeSinkWriter to write SinkRecords to MaxCompute tunnels. This example covers setting up the ODPS client, creating records, configuring the connector, initializing the writer, executing the write operation, and closing the writer for commit.

```java
import com.aliyun.odps.Odps;
import com.aliyun.odps.kafka.connect.MaxComputeSinkWriter;
import com.aliyun.odps.kafka.connect.SinkStatusContext;
import com.aliyun.odps.kafka.connect.converter.RecordConverter;
import com.aliyun.odps.kafka.connect.converter.DefaultRecordConverter;
import com.aliyun.odps.kafka.connect.MaxComputeSinkConnectorConfig;
import org.apache.kafka.connect.sink.SinkRecord;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class WriterExample {
    public void writeRecords() throws Exception {
        // Setup ODPS client
        Odps odps = createOdpsClient();

        // Create records to write
        List<SinkRecord> records = new ArrayList<>();
        for (int i = 0; i < 8000; i++) {
            SinkRecord record = new SinkRecord(
                "events", 0, null, null, null,
                "{\"user\":\"alice\",\"action\":\"login\"}",
                i
            );
            records.add(record);
        }

        // Status context for tracking
        SinkStatusContext statusContext = new SinkStatusContext(8000);

        // Configuration
        MaxComputeSinkConnectorConfig config = createConfig();

        // Record converter
        RecordConverter converter = createConverter(odps, config);

        // Create writer
        MaxComputeSinkWriter writer = new MaxComputeSinkWriter(
            odps,
            records,
            statusContext,
            config,
            "my_project",
            "events_table",
            converter,
            false, // useStreamingTunnel
            null   // errorReporter
        );

        // Execute write (blocking)
        Boolean success = writer.call();
        // Returns true if all writes succeeded
        // Records automatically partitioned by time window (HOUR)
        // Partition format: pt=yyyy-MM-dd HH (if use_new_partition_format=true)

        // Get bytes written
        long bytesWritten = writer.getTotalBytes();
        // Example: 524288 (512KB)

        // Check offset range
        Long minOffset = writer.getMinOffset();
        // Returns null if successfully flushed, else minimum uncommitted offset

        // Close and commit
        writer.close();
        // Commits data to MaxCompute with retry logic
    }

    private Odps createOdpsClient() {
        // Implementation details
        // Replace with actual ODPS account setup
        return new Odps(/* account */);
    }

    private MaxComputeSinkConnectorConfig createConfig() {
        Map<String, String> props = new HashMap<>();
        props.put("endpoint", "http://service.cn-shanghai.maxcompute.aliyun.com/api");
        props.put("project", "my_project");
        props.put("table", "events_table");
        props.put("access_id", "LTAI5txxxxxxxx"); // Replace with your Access ID
        props.put("access_key", "xxxxxxxxxxxxxx"); // Replace with your Access Key
        props.put("tunnel_endpoint", ""); // Optional: specify if needed
        props.put("partition_window_type", "HOUR");
        props.put("use_new_partition_format", "true");
        props.put("buffer_size_kb", "65536");
        props.put("retry_times", "3");
        props.put("skip_error", "false");
        props.put("time_zone", "Asia/Shanghai");
        return new MaxComputeSinkConnectorConfig(props);
    }

    private RecordConverter createConverter(Odps odps, MaxComputeSinkConnectorConfig config) {
        // Implementation details - builds converter based on format/mode
        // This is a placeholder, actual implementation depends on data format and MaxCompute table schema.
        return new DefaultRecordConverter();
    }
}

```

--------------------------------

### MaxComputeSinkConnector Lifecycle and Task Configuration

Source: https://context7.com/aliyun/maxcompute-kafka-connector/llms.txt

Demonstrates the lifecycle of the MaxComputeSinkConnector, including starting the connector with configuration, retrieving the task class, and generating task configurations for parallel processing. It requires Kafka Connect API and MaxCompute connector libraries.

```java
import com.aliyun.odps.kafka.connect.MaxComputeSinkConnector;
import com.aliyun.odps.kafka.connect.MaxComputeSinkConnectorConfig;
import org.apache.kafka.connect.connector.Task;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

// Connector lifecycle
public class ConnectorExample {
    public void demonstrateLifecycle() {
        MaxComputeSinkConnector connector = new MaxComputeSinkConnector();

        // Configuration map
        Map<String, String> config = new HashMap<>();
        config.put("endpoint", "http://service.cn-shanghai.maxcompute.aliyun.com/api");
        config.put("project", "my_project");
        config.put("table", "events");
        config.put("access_id", "LTAI5txxxxxxxx");
        config.put("access_key", "xxxxxxxxxxxxxx");
        config.put("account_type", "ALIYUN");
        config.put("format", "TEXT");
        config.put("mode", "DEFAULT");
        config.put("partition_window_type", "HOUR");

        // Start connector - validates project/table existence
        connector.start(config);

        // Get task class
        Class<? extends Task> taskClass = connector.taskClass();
        // Returns: MaxComputeSinkTask.class

        // Generate task configs for parallel processing
        List<Map<String, String>> taskConfigs = connector.taskConfigs(3);
        // Creates 3 task configurations with identical settings

        // Stop connector
        connector.stop();
    }
}
```

--------------------------------

### Create MaxCompute Kafka Connector with Aliyun Credentials

Source: https://context7.com/aliyun/maxcompute-kafka-connector/llms.txt

This example demonstrates how to create a MaxCompute Kafka Connector instance using Aliyun credentials for authentication. It specifies the Kafka topics to consume, MaxCompute project and table, data format, and performance tuning parameters like buffer size and batch size. The connector is configured for batch processing in DEFAULT mode.

```bash
curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "maxcompute-sink-prod",
    "config": {
      "connector.class": "com.aliyun.odps.kafka.connect.MaxComputeSinkConnector",
      "tasks.max": "3",
      "topics": "user_events,transaction_logs",
      "endpoint": "http://service.cn-shanghai.maxcompute.aliyun.com/api",
      "tunnel_endpoint": "http://dt.cn-shanghai.maxcompute.aliyun.com",
      "project": "production_analytics",
      "table": "kafka_events",
      "account_type": "ALIYUN",
      "access_id": "LTAI5txxxxxxxx",
      "access_key": "xxxxxxxxxxxxxxxxxx",
      "format": "TEXT",
      "mode": "DEFAULT",
      "partition_window_type": "HOUR",
      "use_new_partition_format": true,
      "use_streaming": false,
      "buffer_size_kb": 65536,
      "sink_pool_size": "16",
      "record_batch_size": "8000",
      "runtime.error.topic.name": "maxcompute_errors",
      "runtime.error.topic.bootstrap.servers": "localhost:9092",
      "skip_error": "false"
    }
  }'
```

--------------------------------

### Create MaxCompute Kafka Connector with STS Token Authentication

Source: https://context7.com/aliyun/maxcompute-kafka-connector/llms.txt

This example shows how to configure the MaxCompute Kafka Connector for temporary credential authentication using STS tokens. It includes parameters for STS endpoint, role name, and region ID, enabling automatic token refresh for enhanced security. This configuration is set up for streaming mode and error skipping.

```bash
curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "maxcompute-sts-sink",
    "config": {
      "connector.class": "com.aliyun.odps.kafka.connect.MaxComputeSinkConnector",
      "tasks.max": "5",
      "topics": "sensor_data",
      "endpoint": "http://service.cn-beijing.maxcompute.aliyun.com/api",
      "tunnel_endpoint": "",
      "project": "iot_platform",
      "table": "sensor_readings",
      "account_type": "STS",
      "access_id": "",
      "access_key": "",
      "account_id": "123456789012****",
      "region_id": "cn-beijing",
      "sts.endpoint": "sts.aliyuncs.com",
      "role_name": "AliyunODPSKafkaConnectorRole",
      "client_timeout_ms": "39600000",
      "format": "TEXT",
      "mode": "VALUE",
      "partition_window_type": "MINUTE",
      "use_new_partition_format": true,
      "use_streaming": true,
      "buffer_size_kb": 131072,
      "sink_pool_size": "32",
      "record_batch_size": "10000",
      "skip_error": "true"
    }
  }'
```

--------------------------------

### Maven Build and Deploy MaxCompute Kafka Connector (Bash)

Source: https://context7.com/aliyun/maxcompute-kafka-connector/llms.txt

Provides the bash command to compile the MaxCompute Kafka Connector using Maven and create an uber JAR containing all dependencies. It also shows how to copy the generated JAR file to the Kafka Connect plugin directory for deployment.

```bash
# Clean and package
mvn clean package

# Output files:
# target/kafka-connector-2.1.0.jar
# target/kafka-connector-2.1.0-jar-with-dependencies.jar

# Deploy to Kafka Connect plugin directory
cp target/kafka-connector-2.1.0-jar-with-dependencies.jar \
   /usr/local/share/kafka/plugins/maxcompute-kafka-connector/

```

--------------------------------

### Generate MaxCompute Authentication Accounts (Java)

Source: https://context7.com/aliyun/maxcompute-kafka-connector/llms.txt

Illustrates how to generate Aliyun and STS credentials for MaxCompute using the AccountGenerator interface. It covers using AliyunAccountGenerator for permanent credentials and STSAccountGenerator for temporary, auto-refreshing credentials, as well as the AccountFactory for dynamic account type selection.

```java
import com.aliyun.odps.kafka.connect.account.AccountGenerator;
import com.aliyun.odps.kafka.connect.account.AccountFactory;
import com.aliyun.odps.kafka.connect.account.impl.AliyunAccountGenerator;
import com.aliyun.odps.kafka.connect.account.impl.STSAccountGenerator;
import com.aliyun.odps.account.Account;
import com.aliyun.odps.account.AliyunAccount;
import java.util.HashMap;
import java.util.Map;

public class AuthenticationExample {
    public void generateAccounts() {
        MaxComputeSinkConnectorConfig config = createConfig();

        // Aliyun permanent credentials
        AccountGenerator<AliyunAccount> aliyunGenerator = new AliyunAccountGenerator();
        AliyunAccount aliyunAccount = aliyunGenerator.generate(config);
        // Uses access_id and access_key from config

        // STS temporary credentials with auto-refresh
        AccountGenerator<Account> stsGenerator = new STSAccountGenerator();
        Account stsAccount = stsGenerator.generate(config);
        // Uses account_id, region_id, role_name, sts.endpoint from config
        // Token auto-refreshes based on client_timeout_ms (default: 11 hours)

        // Factory pattern for dynamic selection
        String accountType = config.getString("account_type"); // "ALIYUN" or "STS"
        AccountFactory factory = new AccountFactory();
        Account account = factory.createAccount(accountType, config);
        // Returns appropriate account based on account_type configuration
    }

    private MaxComputeSinkConnectorConfig createConfig() {
        Map<String, String> props = new HashMap<>();
        props.put("account_type", "STS");
        props.put("account_id", "123456789012****");
        props.put("region_id", "cn-shanghai");
        props.put("sts.endpoint", "sts.aliyuncs.com");
        props.put("role_name", "AliyunODPSKafkaRole");
        props.put("client_timeout_ms", "39600000");
        return new MaxComputeSinkConnectorConfig(props);
    }
}

```

--------------------------------

### MaxComputeSinkTask Record Processing and Commit

Source: https://context7.com/aliyun/maxcompute-kafka-connector/llms.txt

Illustrates the MaxComputeSinkTask's role in consuming Kafka SinkRecords, converting them to MaxCompute format, and writing to tunnel endpoints. It includes task startup, partition opening, record buffering, flushing, pre-committing offsets, and stopping the task. Requires Kafka Connect API, MaxCompute connector libraries, and Alibaba Cloud SDKs.

```java
import com.aliyun.odps.kafka.connect.MaxComputeSinkTask;
import org.apache.kafka.connect.sink.SinkRecord;
import org.apache.kafka.connect.sink.SinkOffsetCommitter;
import org.apache.kafka.connect.data.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class TaskExample {
    private MaxComputeSinkTask task;

    public void processRecords() {
        task = new MaxComputeSinkTask();

        // Configuration
        Map<String, String> config = new HashMap<>();
        config.put("endpoint", "http://service.cn-shanghai.maxcompute.aliyun.com/api");
        config.put("project", "analytics");
        config.put("table", "events");
        config.put("access_id", "LTAI5txxxxxxxx");
        config.put("access_key", "xxxxxxxxxxxxxx");
        config.put("account_type", "ALIYUN");
        config.put("tunnel_endpoint", "http://dt.cn-shanghai.maxcompute.aliyun.com");
        config.put("format", "TEXT");
        config.put("mode", "DEFAULT");
        config.put("partition_window_type", "HOUR");
        config.put("use_new_partition_format", "TRUE");
        config.put("sink_pool_size", "16");
        config.put("record_batch_size", "8000");
        config.put("buffer_size_kb", "65536");
        config.put("use_streaming", "FALSE");
        config.put("skip_error", "FALSE");

        // Start task
        task.start(config);

        // Open partitions
        Collection<TopicPartition> partitions = Arrays.asList(
            new TopicPartition("user_events", 0),
            new TopicPartition("user_events", 1)
        );
        task.open(partitions);

        // Process records batch
        Collection<SinkRecord> records = createSampleRecords();
        task.put(records);
        // Records buffered until batch size (8000) reached
        // Then submitted to thread pool for parallel writing

        // Flush and commit
        Map<TopicPartition, OffsetAndMetadata> offsets = getCurrentOffsets();
        task.flush(offsets);
        Map<TopicPartition, OffsetAndMetadata> toCommit = task.preCommit(offsets);
        // Returns safe offsets to commit based on completed writes

        // Close partitions
        task.close(partitions);

        // Stop task
        task.stop();
    }

    private Collection<SinkRecord> createSampleRecords() {
        List<SinkRecord> records = new ArrayList<>();
        for (int i = 0; i < 5000; i++) {
            SinkRecord record = new SinkRecord(
                "user_events", 0, null, null, null,
                "{\"user_id\":123,\"event\":\"click\",\"timestamp\":1634567890}",
                i
            );
            records.add(record);
        }
        return records;
    }

    private Map<TopicPartition, OffsetAndMetadata> getCurrentOffsets() {
        Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
        offsets.put(new TopicPartition("user_events", 0), new OffsetAndMetadata(5000));
        return offsets;
    }
}
```

--------------------------------

### Configure MaxCompute Sink Connector in Java

Source: https://context7.com/aliyun/maxcompute-kafka-connector/llms.txt

This snippet demonstrates how to initialize and use the MaxComputeSinkConnectorConfig class in Java. It shows how to populate configuration properties, create a config object, access typed configuration values, and retrieve the configuration definition for validation. Dependencies include Kafka Connect configuration classes and MaxCompute connector classes.

```java
import com.aliyun.odps.kafka.connect.MaxComputeSinkConnectorConfig;
import com.aliyun.odps.kafka.connect.MaxComputeSinkConnectorConfig.BaseParameter;
import org.apache.kafka.common.config.ConfigDef;

import java.util.HashMap;
import java.util.Map;

public class ConfigExample {
    public void configureConnector() {
        // Create config from properties
        Map<String, String> props = new HashMap<>();
        props.put(BaseParameter.MAXCOMPUTE_ENDPOINT.getName(),
                  "http://service.cn-shanghai.maxcompute.aliyun.com/api");
        props.put(BaseParameter.MAXCOMPUTE_PROJECT.getName(), "my_project");
        props.put(BaseParameter.MAXCOMPUTE_TABLE.getName(), "kafka_sink");
        props.put(BaseParameter.ACCESS_ID.getName(), "LTAI5txxxxxxxx");
        props.put(BaseParameter.ACCESS_KEY.getName(), "xxxxxxxxxxxxxx");
        props.put(BaseParameter.ACCOUNT_TYPE.getName(), "ALIYUN");
        props.put(BaseParameter.FORMAT.getName(), "TEXT");
        props.put(BaseParameter.MODE.getName(), "DEFAULT");
        props.put(BaseParameter.PARTITION_WINDOW_TYPE.getName(), "HOUR");
        props.put(BaseParameter.USE_NEW_PARTITION_FORMAT.getName(), "true");
        props.put(BaseParameter.USE_STREAM_TUNNEL.getName(), "false");
        props.put(BaseParameter.BUFFER_SIZE_KB.getName(), "65536");
        props.put(BaseParameter.POOL_SIZE.getName(), "16");
        props.put(BaseParameter.RECORD_BATCH_SIZE.getName(), "8000");
        props.put(BaseParameter.SKIP_ERROR.getName(), "false");
        props.put(BaseParameter.CLIENT_TIMEOUT_MS.getName(), "39600000"); // 11 hours

        MaxComputeSinkConnectorConfig config = new MaxComputeSinkConnectorConfig(props);

        // Access typed values
        String endpoint = config.getString(BaseParameter.MAXCOMPUTE_ENDPOINT.getName());
        // "http://service.cn-shanghai.maxcompute.aliyun.com/api"

        int poolSize = config.getInt(BaseParameter.POOL_SIZE.getName());
        // 16

        boolean useStreaming = config.getBoolean(BaseParameter.USE_STREAM_TUNNEL.getName());
        // false

        long timeout = config.getLong(BaseParameter.CLIENT_TIMEOUT_MS.getName());
        // 39600000

        // Get config definition for validation
        ConfigDef configDef = MaxComputeSinkConnectorConfig.conf();
        // Returns ConfigDef with all parameter definitions and constraints
    }
}
```

--------------------------------

### Manage Kafka Connectors via REST API

Source: https://context7.com/aliyun/maxcompute-kafka-connector/llms.txt

Demonstrates common Kafka Connect REST API operations for managing connectors, such as listing, checking status, pausing, resuming, restarting, updating configurations, and deleting specific connectors. These commands interact with the Kafka Connect worker's HTTP interface.

```bash
# List all connectors
curl http://localhost:8083/connectors
```

```bash
# Check connector status
curl http://localhost:8083/connectors/maxcompute-sink-prod/status
```

```bash
# Pause connector
curl -X PUT http://localhost:8083/connectors/maxcompute-sink-prod/pause
```

```bash
# Resume connector
curl -X PUT http://localhost:8083/connectors/maxcompute-sink-prod/resume
```

```bash
# Restart connector
curl -X POST http://localhost:8083/connectors/maxcompute-sink-prod/restart
```

```bash
# Restart specific task
curl -X POST http://localhost:8083/connectors/maxcompute-sink-prod/tasks/0/restart
```

```bash
# Update configuration
curl -X PUT http://localhost:8083/connectors/maxcompute-sink-prod/config \
  -H "Content-Type: application/json" \
  -d '{
    "connector.class": "com.aliyun.odps.kafka.connect.MaxComputeSinkConnector",
    "tasks.max": "5",
    "topics": "user_events",
    "endpoint": "http://service.cn-shanghai.maxcompute.aliyun.com/api",
    "project": "production_analytics",
    "table": "kafka_events",
    "sink_pool_size": "32"
  }'
```

```bash
# Delete connector
curl -X DELETE http://localhost:8083/connectors/maxcompute-sink-prod
```

--------------------------------

### Convert Kafka Records to MaxCompute Format (Java)

Source: https://context7.com/aliyun/maxcompute-kafka-connector/llms.txt

Demonstrates how to use the RecordConverter interface to transform Kafka SinkRecord objects into MaxCompute Record format. It shows how to build a converter with specific formats (TEXT, CSV, BINARY) and modes (DEFAULT, KEY, VALUE), and how to use it to convert a SinkRecord.

```java
import com.aliyun.odps.kafka.connect.converter.RecordConverter;
import com.aliyun.odps.kafka.connect.converter.RecordConverterBuilder;
import org.apache.kafka.connect.sink.SinkRecord;
import com.aliyun.odps.data.Record;
import com.aliyun.odps.TableSchema;
import java.io.IOException;

public class ConverterExample {
    public void convertRecords() throws IOException {
        // Build converter
        RecordConverterBuilder builder = new RecordConverterBuilder();
        builder.format(RecordConverterBuilder.Format.TEXT)
               .mode(RecordConverterBuilder.Mode.DEFAULT)
               .schema(getTableSchema());

        RecordConverter converter = builder.build();

        // Create Kafka record
        SinkRecord sinkRecord = new SinkRecord(
            "user_events",        // topic
            0,                    // partition
            null,                 // keySchema
            "user123",            // key
            null,                 // valueSchema
            "{\"event\":\"click\",\"timestamp\":1634567890,\"url\":\"/home\"}",  // value
            12345                 // offset
        );

        // Create MaxCompute record (reusable)
        Record mcRecord = createEmptyRecord();

        // Convert
        converter.convert(sinkRecord, mcRecord);

        // mcRecord now contains:
        // - topic: "user_events"
        // - partition: 0
        // - offset: 12345
        // - key: "user123"
        // - value: "{\"event\":\"click\",\"timestamp\":1634567890,\"url\":\"/home\"}"
        // - insert_time: <current timestamp>

        // Different formats available:
        // Format.TEXT - stores entire value as string
        // Format.CSV - parses CSV and maps to columns
        // Format.BINARY - stores raw bytes

        // Different modes available:
        // Mode.DEFAULT - writes both key and value columns
        // Mode.KEY - writes only key data to value column
        // Mode.VALUE - writes only value data (ignores key)
    }

    private TableSchema getTableSchema() {
        // Return MaxCompute table schema
        return null;
    }

    private Record createEmptyRecord() {
        // Return empty MaxCompute record
        return null;
    }
}

```

--------------------------------

### Connector Management API

Source: https://context7.com/aliyun/maxcompute-kafka-connector/llms.txt

Manage the lifecycle of MaxCompute Kafka Connectors using the Kafka Connect REST API.

```APIDOC
## GET /connectors

### Description
Lists all available connectors in the Kafka Connect cluster.

### Method
GET

### Endpoint
/connectors

### Parameters
None

### Request Example
```bash
curl http://localhost:8083/connectors
```

### Response
#### Success Response (200)
- **connectors** (array) - An array of connector names.

#### Response Example
```json
["maxcompute-sink-prod", "another-connector"]
```
```

```APIDOC
## GET /connectors/{connectorName}/status

### Description
Retrieves the status of a specific connector, including its tasks.

### Method
GET

### Endpoint
/connectors/{connectorName}/status

### Parameters
#### Path Parameters
- **connectorName** (string) - Required - The name of the connector.

### Request Example
```bash
curl http://localhost:8083/connectors/maxcompute-sink-prod/status
```

### Response
#### Success Response (200)
- **name** (string) - The name of the connector.
- **connector** (object) - Details about the connector.
  - **state** (string) - The current state of the connector (e.g., RUNNING, PAUSED, FAILED).
  - **worker_version** (string) - The version of the Kafka Connect worker.
- **tasks** (array) - An array of task statuses.
  - **id** (object) - The ID of the task.
    - **connector** (string) - The name of the connector the task belongs to.
    - **task** (integer) - The index of the task.
  - **state** (string) - The current state of the task (e.g., RUNNING, PAUSED, FAILED).
  - **worker_version** (string) - The version of the Kafka Connect worker.

#### Response Example
```json
{
  "name": "maxcompute-sink-prod",
  "connector": {
    "state": "RUNNING",
    "worker_version": "2.8.0"
  },
  "tasks": [
    {
      "id": {
        "connector": "maxcompute-sink-prod",
        "task": 0
      },
      "state": "RUNNING",
      "worker_version": "2.8.0"
    }
  ]
}
```
```

```APIDOC
## PUT /connectors/{connectorName}/pause

### Description
Pauses a connector, stopping all its tasks from processing data.

### Method
PUT

### Endpoint
/connectors/{connectorName}/pause

### Parameters
#### Path Parameters
- **connectorName** (string) - Required - The name of the connector to pause.

### Request Example
```bash
curl -X PUT http://localhost:8083/connectors/maxcompute-sink-prod/pause
```

### Response
#### Success Response (200)
- Returns the configuration of the connector when it was paused.

#### Response Example
```json
{
  "name": "maxcompute-sink-prod",
  "config": {
    "connector.class": "com.aliyun.odps.kafka.connect.MaxComputeSinkConnector",
    "tasks.max": "5",
    "topics": "user_events",
    "endpoint": "http://service.cn-shanghai.maxcompute.aliyun.com/api",
    "project": "production_analytics",
    "table": "kafka_events",
    "sink_pool_size": "32"
  },
  "tasks": [
    {
      "connector": "maxcompute-sink-prod",
      "task": 0
    }
  ],
  "type": "sink"
}
```
```

```APIDOC
## PUT /connectors/{connectorName}/resume

### Description
Resumes a paused connector, allowing its tasks to resume data processing.

### Method
PUT

### Endpoint
/connectors/{connectorName}/resume

### Parameters
#### Path Parameters
- **connectorName** (string) - Required - The name of the connector to resume.

### Request Example
```bash
curl -X PUT http://localhost:8083/connectors/maxcompute-sink-prod/resume
```

### Response
#### Success Response (200)
- Returns the configuration of the connector when it was resumed.

#### Response Example
```json
{
  "name": "maxcompute-sink-prod",
  "config": {
    "connector.class": "com.aliyun.odps.kafka.connect.MaxComputeSinkConnector",
    "tasks.max": "5",
    "topics": "user_events",
    "endpoint": "http://service.cn-shanghai.maxcompute.aliyun.com/api",
    "project": "production_analytics",
    "table": "kafka_events",
    "sink_pool_size": "32"
  },
  "tasks": [
    {
      "connector": "maxcompute-sink-prod",
      "task": 0
    }
  ],
  "type": "sink"
}
```
```

```APIDOC
## POST /connectors/{connectorName}/restart

### Description
Restarts a connector and all of its tasks. This is useful for applying configuration changes or recovering from certain failures.

### Method
POST

### Endpoint
/connectors/{connectorName}/restart

### Parameters
#### Path Parameters
- **connectorName** (string) - Required - The name of the connector to restart.

### Request Example
```bash
curl -X POST http://localhost:8083/connectors/maxcompute-sink-prod/restart
```

### Response
#### Success Response (200)
- Returns the configuration of the connector after restart.

#### Response Example
```json
{
  "name": "maxcompute-sink-prod",
  "config": {
    "connector.class": "com.aliyun.odps.kafka.connect.MaxComputeSinkConnector",
    "tasks.max": "5",
    "topics": "user_events",
    "endpoint": "http://service.cn-shanghai.maxcompute.aliyun.com/api",
    "project": "production_analytics",
    "table": "kafka_events",
    "sink_pool_size": "32"
  },
  "tasks": [
    {
      "connector": "maxcompute-sink-prod",
      "task": 0
    }
  ],
  "type": "sink"
}
```
```

```APIDOC
## POST /connectors/{connectorName}/tasks/{taskId}/restart

### Description
Restarts a specific task within a connector. This can be used to recover a failed task without restarting the entire connector.

### Method
POST

### Endpoint
/connectors/{connectorName}/tasks/{taskId}/restart

### Parameters
#### Path Parameters
- **connectorName** (string) - Required - The name of the connector.
- **taskId** (integer) - Required - The ID of the task to restart.

### Request Example
```bash
curl -X POST http://localhost:8083/connectors/maxcompute-sink-prod/tasks/0/restart
```

### Response
#### Success Response (200)
- Returns the status of the restarted task.

#### Response Example
```json
{
  "id": {
    "connector": "maxcompute-sink-prod",
    "task": 0
  },
  "state": "RUNNING",
  "worker_version": "2.8.0"
}
```
```

```APIDOC
## PUT /connectors/{connectorName}/config

### Description
Updates the configuration of an existing connector. Note that some configuration changes may require the connector to be restarted to take effect.

### Method
PUT

### Endpoint
/connectors/{connectorName}/config

### Parameters
#### Path Parameters
- **connectorName** (string) - Required - The name of the connector whose configuration to update.

#### Request Body
- **config** (object) - Required - The new configuration for the connector.
  - **connector.class** (string) - Required - The Java class name of the connector.
  - **tasks.max** (string) - Required - The maximum number of tasks for the connector.
  - **topics** (string) - Required - The Kafka topics to consume from.
  - **endpoint** (string) - Required - The MaxCompute service endpoint.
  - **project** (string) - Required - The MaxCompute project name.
  - **table** (string) - Required - The MaxCompute table name.
  - **sink_pool_size** (string) - Optional - The size of the sink thread pool.

### Request Example
```json
{
  "config": {
    "connector.class": "com.aliyun.odps.kafka.connect.MaxComputeSinkConnector",
    "tasks.max": "5",
    "topics": "user_events",
    "endpoint": "http://service.cn-shanghai.maxcompute.aliyun.com/api",
    "project": "production_analytics",
    "table": "kafka_events",
    "sink_pool_size": "32"
  }
}
```

### Response
#### Success Response (200)
- Returns the updated configuration of the connector.

#### Response Example
```json
{
  "name": "maxcompute-sink-prod",
  "config": {
    "connector.class": "com.aliyun.odps.kafka.connect.MaxComputeSinkConnector",
    "tasks.max": "5",
    "topics": "user_events",
    "endpoint": "http://service.cn-shanghai.maxcompute.aliyun.com/api",
    "project": "production_analytics",
    "table": "kafka_events",
    "sink_pool_size": "32"
  },
  "tasks": [
    {
      "connector": "maxcompute-sink-prod",
      "task": 0
    }
  ],
  "type": "sink"
}
```
```

```APIDOC
## DELETE /connectors/{connectorName}

### Description
Deletes a connector and all of its tasks. This operation stops the connector from processing data and removes it from the Kafka Connect cluster.

### Method
DELETE

### Endpoint
/connectors/{connectorName}

### Parameters
#### Path Parameters
- **connectorName** (string) - Required - The name of the connector to delete.

### Request Example
```bash
curl -X DELETE http://localhost:8083/connectors/maxcompute-sink-prod
```

### Response
#### Success Response (200)
- Returns an empty JSON object upon successful deletion.

#### Response Example
```json
{}
```
```

--------------------------------

### Restart Kafka Connect Service

Source: https://context7.com/aliyun/maxcompute-kafka-connector/llms.txt

Restarts the entire Kafka Connect service using systemctl. This is a system-level operation affecting all connectors running on the worker.

```bash
systemctl restart confluent-kafka-connect
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.