CryoThrust commented on code in PR #12684:
URL: https://github.com/apache/seatunnel/pull/12684#discussion_r4214539681
##########
seatunnel-connectors-v2/connector-iotdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/iotdb/sink/IoTDBSinkClient.java:
##########
@@ -104,68 +167,230 @@ public synchronized void write(IoTDBRecord record)
throws IOException {
public synchronized void close() throws IOException {
flush();
- try {
- if (session != null) {
- session.close();
+ for (int i = 0; i < sessions.length; i++) {
+ if (sessions[i] != null) {
+ try {
+ sessions[i].close();
+ } catch (IoTDBConnectionException e) {
+ log.error("Close IoTDB client failed.", e);
+ throw new IotdbConnectorException(
+ IotdbConnectorErrorCode.CLOSE_CLIENT_FAILED,
+ "Close IoTDB client failed.",
+ e);
+ }
}
- } catch (IoTDBConnectionException e) {
- log.error("Close IoTDB client failed.", e);
- throw new IotdbConnectorException(
- IotdbConnectorErrorCode.CLOSE_CLIENT_FAILED, "Close IoTDB
client failed.", e);
}
}
+ /**
+ * Writes are routed per device to the node that last accepted them (the
region leader). A
+ * REDIRECTION_RECOMMEND reply means the current node is no longer the
leader for those devices:
+ * the affected devices are re-routed to another node and retried
immediately, without waiting.
+ * Only network-level failures go through backoff waiting.
+ */
synchronized void flush() throws IOException {
checkFlushException();
if (batchList.isEmpty()) {
return;
}
- BatchRecords batchRecords = new BatchRecords(batchList);
- for (int i = 0; i <= sinkConfig.getMaxRetries(); i++) {
- try {
- if (batchRecords.getTypesList().isEmpty()) {
- session.insertRecords(
- batchRecords.getDeviceIds(),
- batchRecords.getTimestamps(),
- batchRecords.getMeasurementsList(),
- batchRecords.getStringValuesList());
+ List<IoTDBRecord> remaining = new ArrayList<>(batchList);
+ Exception[] lastError = new Exception[1];
+ int nodeCount = sessions.length;
+ // Redirects rotate endpoints without consuming the retry budget; the
round bound only
+ // exists so a leaderless or broken cluster can never spin this loop
forever.
+ int maxRounds = nodeCount + retryBudget() + 1;
Review Comment:
`maxRounds` is also the effective limit for network failures: `writeGroup()`
requeues a failed group on the same endpoint without checking a separate
network budget. With 3 `node_urls`, `max_retries=0`, backoff disabled, and the
first node always throwing `IoTDBConnectionException`, my added test observes 4
insert calls instead of 1. The previous implementation allowed `max_retries +
1` calls. Please keep redirect traversal bounded separately from network
retries, so adding endpoints does not increase the configured network retry
limit. A test should cover both zero and nonzero retries; redirect-only
traversal should not consume that network budget.
##########
seatunnel-connectors-v2/connector-iotdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/iotdb/sink/IoTDBSinkClient.java:
##########
@@ -104,68 +167,230 @@ public synchronized void write(IoTDBRecord record)
throws IOException {
public synchronized void close() throws IOException {
flush();
- try {
- if (session != null) {
- session.close();
+ for (int i = 0; i < sessions.length; i++) {
+ if (sessions[i] != null) {
+ try {
+ sessions[i].close();
+ } catch (IoTDBConnectionException e) {
+ log.error("Close IoTDB client failed.", e);
+ throw new IotdbConnectorException(
+ IotdbConnectorErrorCode.CLOSE_CLIENT_FAILED,
+ "Close IoTDB client failed.",
+ e);
Review Comment:
Throwing here stops cleanup of later sessions. I create n0 (follower) and n1
(leader), successfully flush a record to n1, then make n0.close() throw: n0 is
closed once, but n1.close() is never attempted. Please attempt all created
sessions even if an earlier close fails, clear the session slots, and retain
the first exception with subsequent errors suppressed. Also make cleanup
unconditional when the preceding `flush()` fails; another focused test observes
zero close calls in that case. The flush-before-cleanup issue already existed
with the single session, whereas skipping later sessions is introduced by this
loop.
--
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]
