Copilot commented on code in PR #101: URL: https://github.com/apache/doris-kafka-connector/pull/101#discussion_r3795765937
########## src/main/java/org/apache/doris/kafka/connector/writer/s3/S3TvfLoad.java: ########## @@ -0,0 +1,244 @@ +/* + * 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.doris.kafka.connector.writer.s3; + +import static org.apache.doris.kafka.connector.writer.s3.TvfSqlUtils.quoteIdentifier; +import static org.apache.doris.kafka.connector.writer.s3.TvfSqlUtils.quoteLiteral; + +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.regex.Pattern; +import org.apache.doris.kafka.connector.cfg.DorisOptions; +import org.apache.doris.kafka.connector.connection.ConnectionProvider; +import org.apache.doris.kafka.connector.exception.DorisException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** Executes S3 TVF INSERT statements and reconciles ambiguous Label results. */ +public class S3TvfLoad { + private static final Logger LOG = LoggerFactory.getLogger(S3TvfLoad.class); + private static final Pattern SESSION_VARIABLE = Pattern.compile("[A-Za-z_][A-Za-z0-9_]*"); + + private final ConnectionProvider connectionProvider; + private final S3TvfSqlBuilder sqlBuilder; + private final String database; + private final String table; + private final List<String> columns; + private final boolean deleteSignEnabled; + private final Map<String, String> sessionVariables; + private final int maxRetries; + + public S3TvfLoad( + ConnectionProvider connectionProvider, + DorisOptions options, + String database, + String table) { + this( + connectionProvider, + new S3TvfSqlBuilder(options.getS3TvfOptions()), + database, + table, + options.getTvfColumns(), + options.isEnableDelete(), + options.getSessionVariables(), + options.getMaxRetries()); + } + + S3TvfLoad( + ConnectionProvider connectionProvider, + S3TvfSqlBuilder sqlBuilder, + String database, + String table, + List<String> columns, + boolean deleteSignEnabled, + Map<String, String> sessionVariables, + int maxRetries) { + this.connectionProvider = connectionProvider; + this.sqlBuilder = sqlBuilder; + this.database = database; + this.table = table; + this.columns = Collections.unmodifiableList(columns); + this.deleteSignEnabled = deleteSignEnabled; + this.sessionVariables = Collections.unmodifiableMap(new LinkedHashMap<>(sessionVariables)); + this.maxRetries = maxRetries; + } + + public void load(String label, List<String> objectKeys) { + String sql = + sqlBuilder.buildInsertSql( + database, table, label, objectKeys, columns, deleteSignEnabled); + for (int attempt = 0; attempt <= maxRetries; attempt++) { + try { + executeInsert(sql); + LOG.info("S3 TVF load committed with label {}", label); + return; + } catch (SQLException e) { Review Comment: These retries run back-to-back and never honor the connector's `retry.interval.ms` setting. For a transient JDBC/Doris outage, all configured attempts can be exhausted immediately, so the new retry handling provides no recovery window and can hammer the service. Pass the configured retry interval into this loader and wait between attempts (while preserving interruption). ########## src/main/java/org/apache/doris/kafka/connector/writer/AsyncS3TvfWriter.java: ########## @@ -0,0 +1,336 @@ +/* + * 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.doris.kafka.connector.writer; + +import java.io.ByteArrayOutputStream; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.UUID; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Supplier; +import org.apache.doris.kafka.connector.cfg.DorisOptions; +import org.apache.doris.kafka.connector.cfg.S3TvfOptions; +import org.apache.doris.kafka.connector.connection.ConnectionProvider; +import org.apache.doris.kafka.connector.converter.RecordService; +import org.apache.doris.kafka.connector.exception.DorisException; +import org.apache.doris.kafka.connector.metrics.DorisConnectMonitor; +import org.apache.doris.kafka.connector.service.DorisSystemService; +import org.apache.doris.kafka.connector.writer.load.DefaultThreadFactory; +import org.apache.doris.kafka.connector.writer.s3.S3ClientObjectStore; +import org.apache.doris.kafka.connector.writer.s3.S3ObjectStore; +import org.apache.doris.kafka.connector.writer.s3.S3TvfLoad; +import org.apache.doris.kafka.connector.writer.s3.S3TvfRecordSerializer; +import org.apache.kafka.connect.sink.SinkRecord; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** Stages combined Kafka records as JSON Lines files and commits them through the S3 TVF. */ +public class AsyncS3TvfWriter extends DorisWriter { + private static final Logger LOG = LoggerFactory.getLogger(AsyncS3TvfWriter.class); + private static final byte NEW_LINE = '\n'; + private static final int UPLOAD_QUEUE_SIZE = 1; + private static final Runnable UPLOAD_BARRIER = () -> {}; + + private S3ObjectStore objectStore; + private S3TvfLoad load; + private S3TvfRecordSerializer serializer; + private Supplier<String> batchUuidSupplier; + private ExecutorService uploadExecutor; + private S3TvfOptions s3Options; + private String normalizedLabelPrefix; + private String normalizedTable; + + private final ByteArrayOutputStream tvfBuffer = new ByteArrayOutputStream(); + private final BlockingQueue<Runnable> uploadQueue = + new LinkedBlockingQueue<>(UPLOAD_QUEUE_SIZE); + private final AtomicReference<DorisException> uploadException = new AtomicReference<>(); + private final List<String> uploadedObjectKeys = new ArrayList<>(); + private int bufferedRecords; + private String batchUuid; + private int fileNumber; + + public AsyncS3TvfWriter( + String tableName, + String topic, + int partition, + DorisOptions dorisOptions, + ConnectionProvider connectionProvider, + DorisSystemService dorisSystemService, + DorisConnectMonitor connectMonitor) { + super( + tableName, + topic, + partition, + dorisOptions, + connectionProvider, + dorisSystemService, + connectMonitor); + initialize( + new S3ClientObjectStore(dorisOptions.getS3TvfOptions()), + new S3TvfLoad(connectionProvider, dorisOptions, dbName, this.tableName), + this.recordService, + () -> UUID.randomUUID().toString(), + Executors.newSingleThreadExecutor( + new DefaultThreadFactory("s3-tvf-upload-" + dorisOptions.getTaskId()))); + } + + AsyncS3TvfWriter( + String tableName, + String topic, + int partition, + DorisOptions dorisOptions, + ConnectionProvider connectionProvider, + DorisSystemService dorisSystemService, + DorisConnectMonitor connectMonitor, + RecordService recordService, + S3ObjectStore objectStore, + S3TvfLoad load, + Supplier<String> batchUuidSupplier, + ExecutorService uploadExecutor) { + super( + tableName, + topic, + partition, + dorisOptions, + connectionProvider, + dorisSystemService, + connectMonitor); + initialize(objectStore, load, recordService, batchUuidSupplier, uploadExecutor); + } + + private void initialize( + S3ObjectStore objectStore, + S3TvfLoad load, + RecordService recordService, + Supplier<String> batchUuidSupplier, + ExecutorService uploadExecutor) { + this.objectStore = objectStore; + this.load = load; + this.recordService = recordService; + this.batchUuidSupplier = batchUuidSupplier; + this.uploadExecutor = uploadExecutor; + this.s3Options = dorisOptions.getS3TvfOptions(); + this.serializer = + new S3TvfRecordSerializer( + dorisOptions.getTvfColumns(), dorisOptions.isEnableDelete()); + this.normalizedLabelPrefix = normalize(dorisOptions.getLabelPrefix()); + this.normalizedTable = normalize(tableIdentifier); + this.uploadExecutor.execute(this::runUploadLoop); + } + + @Override + public synchronized void insert(SinkRecord record) { + checkUploadException(); + String processedRecord = recordService.getProcessedRecord(record); + if (processedRecord == null) { + return; + } + String serialized = serializer.serialize(processedRecord); + if (serialized.isEmpty()) { + return; + } + byte[] bytes = serialized.getBytes(StandardCharsets.UTF_8); + int bytesWithNewLine = bytes.length + 1; + tvfBuffer.write(bytes, 0, bytes.length); + tvfBuffer.write(NEW_LINE); + bufferedRecords++; + connectMonitor.addAndGetBuffMemoryUsage(bytesWithNewLine); + + if (tvfBuffer.size() >= dorisOptions.getFileSize() + || (dorisOptions.getRecordNum() != 0 + && bufferedRecords >= dorisOptions.getRecordNum())) { + submitBuffer(); + } + } + + @Override + public synchronized void flushBuffer() { + submitBuffer(); + } + + @Override + public synchronized void commitFlush() { + submitBuffer(); + if (batchUuid == null) { + return; + } + + waitForUploads(); + List<String> objectKeys = Collections.unmodifiableList(new ArrayList<>(uploadedObjectKeys)); + String label = buildLabel(); + load.load(label, objectKeys); + uploadedObjectKeys.clear(); Review Comment: After a load attempt throws, this writer leaves the batch UUID and uploaded-key list active. A later `insert()` can therefore append files to the same label before `commitFlush()` is retried. If the original INSERT actually finished but its response was lost, label reconciliation reports `FINISHED` and these newly appended files are cleared without ever being loaded. Freeze the failed commit snapshot and reject/route subsequent inserts until that exact snapshot has been reconciled. ########## src/main/java/org/apache/doris/kafka/connector/writer/AsyncS3TvfWriter.java: ########## @@ -0,0 +1,336 @@ +/* + * 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.doris.kafka.connector.writer; + +import java.io.ByteArrayOutputStream; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.UUID; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Supplier; +import org.apache.doris.kafka.connector.cfg.DorisOptions; +import org.apache.doris.kafka.connector.cfg.S3TvfOptions; +import org.apache.doris.kafka.connector.connection.ConnectionProvider; +import org.apache.doris.kafka.connector.converter.RecordService; +import org.apache.doris.kafka.connector.exception.DorisException; +import org.apache.doris.kafka.connector.metrics.DorisConnectMonitor; +import org.apache.doris.kafka.connector.service.DorisSystemService; +import org.apache.doris.kafka.connector.writer.load.DefaultThreadFactory; +import org.apache.doris.kafka.connector.writer.s3.S3ClientObjectStore; +import org.apache.doris.kafka.connector.writer.s3.S3ObjectStore; +import org.apache.doris.kafka.connector.writer.s3.S3TvfLoad; +import org.apache.doris.kafka.connector.writer.s3.S3TvfRecordSerializer; +import org.apache.kafka.connect.sink.SinkRecord; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** Stages combined Kafka records as JSON Lines files and commits them through the S3 TVF. */ +public class AsyncS3TvfWriter extends DorisWriter { + private static final Logger LOG = LoggerFactory.getLogger(AsyncS3TvfWriter.class); + private static final byte NEW_LINE = '\n'; + private static final int UPLOAD_QUEUE_SIZE = 1; + private static final Runnable UPLOAD_BARRIER = () -> {}; + + private S3ObjectStore objectStore; + private S3TvfLoad load; + private S3TvfRecordSerializer serializer; + private Supplier<String> batchUuidSupplier; + private ExecutorService uploadExecutor; + private S3TvfOptions s3Options; + private String normalizedLabelPrefix; + private String normalizedTable; + + private final ByteArrayOutputStream tvfBuffer = new ByteArrayOutputStream(); + private final BlockingQueue<Runnable> uploadQueue = + new LinkedBlockingQueue<>(UPLOAD_QUEUE_SIZE); + private final AtomicReference<DorisException> uploadException = new AtomicReference<>(); + private final List<String> uploadedObjectKeys = new ArrayList<>(); + private int bufferedRecords; + private String batchUuid; + private int fileNumber; + + public AsyncS3TvfWriter( + String tableName, + String topic, + int partition, + DorisOptions dorisOptions, + ConnectionProvider connectionProvider, + DorisSystemService dorisSystemService, + DorisConnectMonitor connectMonitor) { + super( + tableName, + topic, + partition, + dorisOptions, + connectionProvider, + dorisSystemService, + connectMonitor); + initialize( + new S3ClientObjectStore(dorisOptions.getS3TvfOptions()), + new S3TvfLoad(connectionProvider, dorisOptions, dbName, this.tableName), + this.recordService, + () -> UUID.randomUUID().toString(), + Executors.newSingleThreadExecutor( + new DefaultThreadFactory("s3-tvf-upload-" + dorisOptions.getTaskId()))); + } + + AsyncS3TvfWriter( + String tableName, + String topic, + int partition, + DorisOptions dorisOptions, + ConnectionProvider connectionProvider, + DorisSystemService dorisSystemService, + DorisConnectMonitor connectMonitor, + RecordService recordService, + S3ObjectStore objectStore, + S3TvfLoad load, + Supplier<String> batchUuidSupplier, + ExecutorService uploadExecutor) { + super( + tableName, + topic, + partition, + dorisOptions, + connectionProvider, + dorisSystemService, + connectMonitor); + initialize(objectStore, load, recordService, batchUuidSupplier, uploadExecutor); + } + + private void initialize( + S3ObjectStore objectStore, + S3TvfLoad load, + RecordService recordService, + Supplier<String> batchUuidSupplier, + ExecutorService uploadExecutor) { + this.objectStore = objectStore; + this.load = load; + this.recordService = recordService; + this.batchUuidSupplier = batchUuidSupplier; + this.uploadExecutor = uploadExecutor; + this.s3Options = dorisOptions.getS3TvfOptions(); + this.serializer = + new S3TvfRecordSerializer( + dorisOptions.getTvfColumns(), dorisOptions.isEnableDelete()); + this.normalizedLabelPrefix = normalize(dorisOptions.getLabelPrefix()); + this.normalizedTable = normalize(tableIdentifier); + this.uploadExecutor.execute(this::runUploadLoop); + } + + @Override + public synchronized void insert(SinkRecord record) { + checkUploadException(); + String processedRecord = recordService.getProcessedRecord(record); + if (processedRecord == null) { + return; + } + String serialized = serializer.serialize(processedRecord); + if (serialized.isEmpty()) { + return; + } + byte[] bytes = serialized.getBytes(StandardCharsets.UTF_8); + int bytesWithNewLine = bytes.length + 1; + tvfBuffer.write(bytes, 0, bytes.length); + tvfBuffer.write(NEW_LINE); + bufferedRecords++; + connectMonitor.addAndGetBuffMemoryUsage(bytesWithNewLine); + + if (tvfBuffer.size() >= dorisOptions.getFileSize() + || (dorisOptions.getRecordNum() != 0 + && bufferedRecords >= dorisOptions.getRecordNum())) { + submitBuffer(); + } + } + + @Override + public synchronized void flushBuffer() { + submitBuffer(); + } + + @Override + public synchronized void commitFlush() { + submitBuffer(); + if (batchUuid == null) { + return; + } + + waitForUploads(); + List<String> objectKeys = Collections.unmodifiableList(new ArrayList<>(uploadedObjectKeys)); + String label = buildLabel(); + load.load(label, objectKeys); + uploadedObjectKeys.clear(); + batchUuid = null; + fileNumber = 0; + } + + private void submitBuffer() { + if (tvfBuffer.size() == 0) { + return; + } + ensureBatch(); + int currentFileNumber = fileNumber++; + String label = buildLabel(); + String fileName = + label + "_" + dorisOptions.getTaskId() + "_" + currentFileNumber + ".json"; + String objectKey = buildObjectKey(fileName); + byte[] content = tvfBuffer.toByteArray(); + int recordCount = bufferedRecords; + putUpload( + () -> { + if (uploadException.get() != null) { + return; + } + try { + objectStore.put(objectKey, content); + uploadedObjectKeys.add(objectKey); + } catch (Exception e) { + uploadException.compareAndSet( + null, + new DorisException("Failed to upload S3 TVF file " + objectKey, e)); + } + }); + connectMonitor.updateBufferMetrics(content.length, recordCount); + connectMonitor.addAndGetTotalSizeOfData(content.length); + connectMonitor.addAndGetTotalNumberOfRecord(recordCount); + connectMonitor.resetMemoryUsage(); + tvfBuffer.reset(); + bufferedRecords = 0; + LOG.info( + "Queued S3 TVF file {} for upload ({} bytes, {} records)", + fileName, + content.length, + recordCount); + } + + private void runUploadLoop() { + LOG.info("S3 TVF upload worker started"); + try { + while (!Thread.currentThread().isInterrupted()) { + uploadQueue.take().run(); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + LOG.info("S3 TVF upload worker stopped"); + } + } + + private void waitForUploads() { + for (int i = 0; i < UPLOAD_QUEUE_SIZE + 1; i++) { + putUpload(UPLOAD_BARRIER); + } + } + + private void putUpload(Runnable upload) { + checkUploadException(); + try { + uploadQueue.put(upload); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new DorisException("Interrupted while queuing an S3 TVF upload", e); + } + checkUploadException(); + } + + private void checkUploadException() { + DorisException exception = uploadException.get(); + if (exception != null) { + throw exception; + } + } + + /** Clears a failed upload batch so Kafka Connect can retry the records. */ + public synchronized void resetAfterUploadFailure() { + if (uploadException.getAndSet(null) == null) { + return; + } + uploadQueue.clear(); + uploadedObjectKeys.clear(); Review Comment: Clearing `uploadException` before draining the worker creates a race with an upload that has already been dequeued. That stale lambda can observe the cleared exception at line 204, upload an object from the failed batch, and append its key after these collections were cleared, contaminating the retried batch. Keep the failure latched until the worker has acknowledged/drained all old work, then reset the state and clear the exception last. ########## src/main/java/org/apache/doris/kafka/connector/cfg/DorisOptions.java: ########## @@ -209,6 +221,66 @@ private Properties getStreamLoadDefaultValues() { return properties; } + private S3TvfOptions buildS3TvfOptions(Map<String, String> config) { + if (!LoadModel.TVF.equals(loadModel)) { + return null; + } + return S3TvfOptions.builder() + .setEndpoint(config.get(DorisSinkConnectorConfig.SINK_S3_ENDPOINT)) + .setRegion(config.get(DorisSinkConnectorConfig.SINK_S3_REGION)) + .setBucket(config.get(DorisSinkConnectorConfig.SINK_S3_BUCKET)) + .setPrefix(config.get(DorisSinkConnectorConfig.SINK_S3_PREFIX)) + .setAccessKey(config.get(DorisSinkConnectorConfig.SINK_S3_ACCESS_KEY)) + .setSecretKey(config.get(DorisSinkConnectorConfig.SINK_S3_SECRET_KEY)) + .setPathStyleAccess( + Boolean.parseBoolean( + config.getOrDefault( + DorisSinkConnectorConfig.SINK_S3_PATH_STYLE_ACCESS, + String.valueOf( + DorisSinkConnectorConfig + .SINK_S3_PATH_STYLE_ACCESS_DEFAULT)))) + .build(); + } + + private List<String> resolveTvfColumns() { + if (!LoadModel.TVF.equals(loadModel)) { + return Collections.emptyList(); + } + return TvfColumnUtils.resolveColumns(streamLoadProp.getProperty("columns")); + } + + private Map<String, String> getSessionVariablesFromConfig(Map<String, String> config) { + if (!LoadModel.TVF.equals(loadModel)) { + return Collections.emptyMap(); + } + Set<String> transportProperties = + new HashSet<>( + Arrays.asList( + "format", + "read_json_by_line", + "compress_type", + "columns", + "partial_columns")); + Map<String, String> variables = new HashMap<>(); + for (Map.Entry<String, String> entry : config.entrySet()) { + if (!entry.getKey().startsWith(DorisSinkConnectorConfig.STREAM_LOAD_PROP_PREFIX)) { + continue; + } + String name = + entry.getKey() + .substring(DorisSinkConnectorConfig.STREAM_LOAD_PROP_PREFIX.length()); + if (!transportProperties.contains(name)) { + variables.put(name, entry.getValue()); + } Review Comment: Every `sink.properties.*` entry not in this short transport list is treated as a Doris session variable. Existing Stream Load-only properties such as `max_filter_ratio` or `strict_mode` will consequently produce `SET SESSION` statements for variables that INSERT/TVF does not support, despite passing connector validation. Whitelist supported TVF session variables (or explicitly map equivalents) and reject unsupported properties during validation. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
