This is an automated email from the ASF dual-hosted git repository.
JNSimba pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris-kafka-connector.git
The following commit(s) were added to refs/heads/master by this push:
new df3f48d [Fix] Handle S3 TVF label and load failures (#103)
df3f48d is described below
commit df3f48d8877fc5a2107bbc6e075987f485cdd79a
Author: wudi <[email protected]>
AuthorDate: Fri Aug 28 14:12:17 2026 +0800
[Fix] Handle S3 TVF label and load failures (#103)
Preserve the complete label prefix by falling back to labelPrefix_UUID when
an S3 TVF label exceeds the default limit.
Persist S3 TVF upload and load failures so Kafka Connect applies its
configured retry budget.
Reset failed S3 TVF batches through the existing sink retry initialization.
---
.../service/DorisCombinedSinkService.java | 2 +-
.../kafka/connector/writer/AsyncS3TvfWriter.java | 37 +++++++++-------
.../e2e/doris/DorisContainerServiceImpl.java | 36 ++++++++++-----
.../connector/writer/AsyncS3TvfWriterTest.java | 51 ++++++++++++++++++++--
4 files changed, 94 insertions(+), 32 deletions(-)
diff --git
a/src/main/java/org/apache/doris/kafka/connector/service/DorisCombinedSinkService.java
b/src/main/java/org/apache/doris/kafka/connector/service/DorisCombinedSinkService.java
index 60ed9a7..fe106b4 100644
---
a/src/main/java/org/apache/doris/kafka/connector/service/DorisCombinedSinkService.java
+++
b/src/main/java/org/apache/doris/kafka/connector/service/DorisCombinedSinkService.java
@@ -51,7 +51,7 @@ public class DorisCombinedSinkService extends
DorisDefaultSinkService {
// it needs to be restarted when retrying
((AsyncStreamLoadWriter) wr).start();
} else if (wr instanceof AsyncS3TvfWriter) {
- ((AsyncS3TvfWriter) wr).resetAfterUploadFailure();
+ ((AsyncS3TvfWriter) wr).resetAfterFailure();
}
}
}
diff --git
a/src/main/java/org/apache/doris/kafka/connector/writer/AsyncS3TvfWriter.java
b/src/main/java/org/apache/doris/kafka/connector/writer/AsyncS3TvfWriter.java
index a37378f..15e3866 100644
---
a/src/main/java/org/apache/doris/kafka/connector/writer/AsyncS3TvfWriter.java
+++
b/src/main/java/org/apache/doris/kafka/connector/writer/AsyncS3TvfWriter.java
@@ -51,6 +51,7 @@ import org.slf4j.LoggerFactory;
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 MAX_LABEL_LENGTH = 128;
private static final int UPLOAD_QUEUE_SIZE = 1;
private static final Runnable UPLOAD_BARRIER = () -> {};
@@ -64,7 +65,7 @@ public class AsyncS3TvfWriter extends DorisWriter {
private final ByteArrayOutputStream tvfBuffer = new
ByteArrayOutputStream();
private final BlockingQueue<Runnable> uploadQueue;
- private final AtomicReference<DorisException> uploadException = new
AtomicReference<>();
+ private final AtomicReference<Throwable> exception = new
AtomicReference<>();
private final List<String> uploadedObjectKeys = new ArrayList<>();
private int bufferedRecords;
private String batchUuid;
@@ -167,7 +168,7 @@ public class AsyncS3TvfWriter extends DorisWriter {
@Override
public synchronized void insert(SinkRecord record) {
- checkUploadException();
+ checkException();
String processedRecord = recordService.getProcessedRecord(record);
if (processedRecord == null) {
return;
@@ -207,6 +208,9 @@ public class AsyncS3TvfWriter extends DorisWriter {
String label = buildLabel();
try {
load.load(label, objectKeys);
+ } catch (RuntimeException e) {
+ exception.compareAndSet(null, e);
+ throw e;
} finally {
finishBatch();
}
@@ -226,14 +230,14 @@ public class AsyncS3TvfWriter extends DorisWriter {
int recordCount = bufferedRecords;
putUpload(
() -> {
- if (uploadException.get() != null) {
+ if (exception.get() != null) {
return;
}
try {
objectStore.put(objectKey, content);
uploadedObjectKeys.add(objectKey);
} catch (Exception e) {
- uploadException.compareAndSet(
+ exception.compareAndSet(
null,
new DorisException("Failed to upload S3 TVF
file " + objectKey, e));
}
@@ -271,26 +275,25 @@ public class AsyncS3TvfWriter extends DorisWriter {
}
private void putUpload(Runnable upload) {
- checkUploadException();
+ checkException();
try {
uploadQueue.put(upload);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new DorisException("Interrupted while queuing an S3 TVF
upload", e);
}
- checkUploadException();
+ checkException();
}
- private void checkUploadException() {
- DorisException exception = uploadException.get();
- if (exception != null) {
- throw exception;
+ private void checkException() {
+ if (exception.get() != null) {
+ throw new DorisException(exception.get());
}
}
- /** Clears a failed upload batch so Kafka Connect can retry the records. */
- public synchronized void resetAfterUploadFailure() {
- if (uploadException.get() == null) {
+ /** Clears a failed batch so Kafka Connect can retry the records. */
+ public synchronized void resetAfterFailure() {
+ if (exception.get() == null) {
return;
}
uploadQueue.clear();
@@ -299,8 +302,8 @@ public class AsyncS3TvfWriter extends DorisWriter {
tvfBuffer.reset();
bufferedRecords = 0;
connectMonitor.resetMemoryUsage();
- uploadException.set(null);
- LOG.info("Reset failed S3 TVF upload batch for retry");
+ exception.set(null);
+ LOG.info("Reset failed S3 TVF batch for retry");
}
/**
@@ -338,7 +341,9 @@ public class AsyncS3TvfWriter extends DorisWriter {
}
private String buildLabel() {
- return normalizedLabelPrefix + "_" + normalizedTable + "_" + batchUuid;
+ String suffix = "_" + batchUuid;
+ String label = normalizedLabelPrefix + "_" + normalizedTable + suffix;
+ return label.length() <= MAX_LABEL_LENGTH ? label :
normalizedLabelPrefix + suffix;
}
private String buildObjectKey(String fileName) {
diff --git
a/src/test/java/org/apache/doris/kafka/connector/e2e/doris/DorisContainerServiceImpl.java
b/src/test/java/org/apache/doris/kafka/connector/e2e/doris/DorisContainerServiceImpl.java
index c28d736..8ff19dc 100644
---
a/src/test/java/org/apache/doris/kafka/connector/e2e/doris/DorisContainerServiceImpl.java
+++
b/src/test/java/org/apache/doris/kafka/connector/e2e/doris/DorisContainerServiceImpl.java
@@ -159,19 +159,33 @@ public class DorisContainerServiceImpl implements
DorisContainerService {
DorisContainerServiceImpl.class.getClassLoader());
LOG.info("Try to connect to Doris.");
Thread.currentThread().setContextClassLoader(urlClassLoader);
- try (Connection connection =
- DriverManager.getConnection(
- String.format(JDBC_URL,
dorisContainer.getHost()),
- USERNAME,
- PASSWORD);
- Statement statement = connection.createStatement()) {
- ResultSet resultSet;
- do {
+ Duration timeout = Duration.ofMinutes(5L);
+ long startNanos = System.nanoTime();
+ Throwable lastFailure = null;
+ boolean frontendReady = false;
+ while (System.nanoTime() - startNanos < timeout.toNanos()) {
+ try (Connection connection = getQueryConnection();
+ Statement statement = connection.createStatement();
+ ResultSet resultSet = statement.executeQuery("show
backends")) {
+ frontendReady = true;
+ lastFailure = null;
LOG.info("Waiting for the Backend to start successfully.");
- resultSet = statement.executeQuery("show backends");
- } while (!isBeReady(resultSet, Duration.ofSeconds(1L)));
+ if (isBeReady(resultSet, Duration.ofSeconds(1L))) {
+ LOG.info("Connected to Doris successfully.");
+ return;
+ }
+ } catch (DorisException | SQLException e) {
+ frontendReady = false;
+ lastFailure = e;
+ LOG.info("Waiting for the Frontend to accept connections.");
+ LockSupport.parkNanos(Duration.ofSeconds(1L).toNanos());
+ }
}
- LOG.info("Connected to Doris successfully.");
+ throw new DorisException(
+ String.format(
+ "Doris %s did not become ready within %d seconds.",
+ frontendReady ? "Backend" : "Frontend",
timeout.getSeconds()),
+ lastFailure);
}
private boolean isBeReady(ResultSet rs, Duration duration) throws
SQLException {
diff --git
a/src/test/java/org/apache/doris/kafka/connector/writer/AsyncS3TvfWriterTest.java
b/src/test/java/org/apache/doris/kafka/connector/writer/AsyncS3TvfWriterTest.java
index 21edfe5..f074b44 100644
---
a/src/test/java/org/apache/doris/kafka/connector/writer/AsyncS3TvfWriterTest.java
+++
b/src/test/java/org/apache/doris/kafka/connector/writer/AsyncS3TvfWriterTest.java
@@ -93,6 +93,37 @@ public class AsyncS3TvfWriterTest {
writer.close();
}
+ @Test
+ public void testLabelDoesNotExceedDorisLimit() throws Exception {
+ RecordingObjectStore store = new RecordingObjectStore();
+ S3TvfLoad load = mock(S3TvfLoad.class);
+ RecordService records = mock(RecordService.class);
+ SinkRecord record = TestRecordBuffer.newSinkRecord("ignored", 1);
+ String labelPrefix = "kafka_tvf_1787740455434";
+
when(records.getProcessedRecord(record)).thenReturn("{\"id\":1,\"name\":\"first\"}");
+ AsyncS3TvfWriter writer =
+ new AsyncS3TvfWriter(
+
"regression_test_stress_load_release_kafka_connector.kafka_connector_tvf_dup",
+ "orders-topic",
+ -1,
+ options(1024, 100, labelPrefix),
+ mock(ConnectionProvider.class),
+ mock(DorisSystemService.class),
+ mock(DorisConnectMonitor.class),
+ records,
+ store,
+ load,
+ Executors.newSingleThreadExecutor());
+
+ writer.insert(record);
+ writer.commitFlush();
+
+ ArgumentCaptor<String> label = ArgumentCaptor.forClass(String.class);
+ verify(load).load(label.capture(), anyList());
+ Assert.assertTrue(label.getValue().matches(labelPrefix +
"_[0-9a-f]{32}"));
+ writer.close();
+ }
+
@Test
public void testSuccessfulCommitStartsNewBatch() throws Exception {
RecordingObjectStore store = new RecordingObjectStore();
@@ -242,7 +273,7 @@ public class AsyncS3TvfWriterTest {
// Reset the failed batch before Kafka Connect retries the
same records.
}
- writer.resetAfterUploadFailure();
+ writer.resetAfterFailure();
store.putFailure = null;
writer.insert(record);
writer.commitFlush();
@@ -276,7 +307,7 @@ public class AsyncS3TvfWriterTest {
}
Assert.assertTrue(uploadQueue.secondUploadDequeued.await(5,
TimeUnit.SECONDS));
- Future<?> reset =
resetExecutor.submit(writer::resetAfterUploadFailure);
+ Future<?> reset = resetExecutor.submit(writer::resetAfterFailure);
Assert.assertTrue(uploadQueue.resetStarted.await(5,
TimeUnit.SECONDS));
uploadQueue.continueSecondUpload.countDown();
@@ -308,7 +339,7 @@ public class AsyncS3TvfWriterTest {
}
@Test
- public void testLoadFailureEndsBatchBeforeRetry() throws Exception {
+ public void testLoadFailureRemainsVisibleUntilReset() throws Exception {
RecordingObjectStore store = new RecordingObjectStore();
S3TvfLoad load = mock(S3TvfLoad.class);
doThrow(new
DorisException("failed")).doNothing().when(load).load(anyString(), anyList());
@@ -324,6 +355,13 @@ public class AsyncS3TvfWriterTest {
} catch (DorisException expected) {
// Kafka Connect can replay the records after the failed commit.
}
+ try {
+ writer.insert(record);
+ Assert.fail("Expected load failure to remain visible");
+ } catch (DorisException expected) {
+ // DorisSinkTask handles the persistent failure through its put
retry budget.
+ }
+ writer.resetAfterFailure();
writer.insert(record);
writer.commitFlush();
@@ -366,6 +404,11 @@ public class AsyncS3TvfWriterTest {
}
private static DorisOptions options(int bufferSize, int recordCount)
throws IOException {
+ return options(bufferSize, recordCount, "tvf");
+ }
+
+ private static DorisOptions options(int bufferSize, int recordCount,
String labelPrefix)
+ throws IOException {
InputStream stream =
AsyncS3TvfWriterTest.class
.getClassLoader()
@@ -376,7 +419,7 @@ public class AsyncS3TvfWriterTest {
properties.put("task_id", "7");
properties.put(DorisSinkConnectorConfig.NAME, "connector");
properties.put(DorisSinkConnectorConfig.DORIS_DATABASE, "");
- properties.put(DorisSinkConnectorConfig.LABEL_PREFIX, "tvf");
+ properties.put(DorisSinkConnectorConfig.LABEL_PREFIX, labelPrefix);
properties.put(DorisSinkConnectorConfig.LOAD_MODEL, "tvf");
properties.put(DorisSinkConnectorConfig.ENABLE_COMBINE_FLUSH, "true");
properties.put(DorisSinkConnectorConfig.DELIVERY_GUARANTEE,
"at_least_once");
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
