DanielLeens commented on PR #10680: URL: https://github.com/apache/seatunnel/pull/10680#issuecomment-6368709411
I prepared a concrete conflict-resolution patch for unchanged PR head `0570563e5bf5892908e0776396ed70d09bcfba3c` against `apache/seatunnel:dev` at `14a27ac26a0a71b56722f86cdf8ec2779094ceaf`. There is an important result after resolving: **the entire resolved tree is identical to that pinned `dev` commit** (`git diff --cached --exit-code 14a27ac26a0a71b56722f86cdf8ec2779094ceaf --` passed). There is no remaining unique source fix after retaining the newer upstream implementation and tests. This makes the supersession discussed above directly reproducible. The resolution retains both [#11431](https://github.com/apache/seatunnel/pull/11431) (`46fefa7c611`) and the newer [#12692](https://github.com/apache/seatunnel/pull/12692) (`787380c35a2`), including pending-batch/label retention, explicit label-state confirmation, ABORTED-only replacement labels, interruption handling, the bounded label-state deadline, Publish Timeout handling, and the upstream regression matrix. Replacing those files with this PR's old versions would lose the newer behavior. If the branch needs to be reconciled for comparison, use a clean checkout at the exact PR head, save the diff below as `pr10680-resolve-conflicts.patch`, and run: ```bash git fetch https://github.com/apache/seatunnel.git 14a27ac26a0a71b56722f86cdf8ec2779094ceaf git -c merge.conflictStyle=merge -c rerere.enabled=false merge --no-commit --no-ff 14a27ac26a0a71b56722f86cdf8ec2779094ceaf # The merge is expected to stop with conflicts in the three files below. git apply --check pr10680-resolve-conflicts.patch git apply pr10680-resolve-conflicts.patch git add -- seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/client/StarRocksStreamLoadVisitor.java seatunnel-connectors-v2/connector-starrocks/src/test/java/org/apache/seatunnel/connectors/seatunnel/starrocks/client/StarRocksSinkManagerTest.java seatunnel-connectors-v2/connector-starrocks/src/test/java/org/apache/seatunnel/connectors/seatunnel/starrocks/client/StarRocksStreamLoadVisitorTest.java git diff --cached --exit-code 14a27ac26a0a71b56722f86cdf8ec2779094ceaf -- ``` The patch applies to the conflict-marked working files produced by that exact merge. If the head/base or conflict output differs, re-evaluate the resolution instead of forcing it. The final comparison must be empty; it is evidence that the source work is already upstream. Maintainers can use that result to decide whether this superseded branch needs any further contribution. Validation performed: exact merge/conflict inspection, forward patch applicability, and full resolved-tree equality to pinned `dev`. No local compilation, unit tests, or runtime tests were run, and this does not claim CI passed. The old Build check reports workflow-run detection failure; a contributor push would require fresh CI on its resulting head. <details> <summary>Conflict-marker resolution patch for the pinned merge</summary> ```diff --- a/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/client/StarRocksStreamLoadVisitor.java +++ b/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/client/StarRocksStreamLoadVisitor.java @@ -67,17 +67,6 @@ private final long labelStateTimeoutMs; public StarRocksStreamLoadVisitor(SinkConfig sinkConfig, TableSchema tableSchema) { -<<<<<<< HEAD - this(sinkConfig, tableSchema, new HttpHelper(sinkConfig)); - } - - /** Creates a visitor with an explicit HTTP boundary for deterministic response-state tests. */ - StarRocksStreamLoadVisitor( - SinkConfig sinkConfig, TableSchema tableSchema, HttpHelper httpHelper) { - this.sinkConfig = sinkConfig; - this.tableSchema = tableSchema; - this.httpHelper = httpHelper; -======= this(sinkConfig, tableSchema, new HttpHelper(sinkConfig), 0L); } @@ -114,7 +103,6 @@ : DEFAULT_LABEL_STATE_TIMEOUT_MS; } this.labelStateTimeoutMs = Math.max(1, configuredTimeoutMs); ->>>>>>> 14a27ac26a0a71b56722f86cdf8ec2779094ceaf checkBatchMaxBytes(sinkConfig.getBatchMaxBytes(), sinkConfig.getBatchMaxSize()); } @@ -185,27 +173,8 @@ errorBuilder.append(JsonUtils.toJsonString(loadResult)); errorBuilder.append('\n'); } - String message = - loadResult.containsKey("Message") - ? String.valueOf(loadResult.get("Message")) - : ""; - boolean labelAlreadyUsed = message.contains("has already been used"); - if (labelAlreadyUsed) { - // A reused label can belong to a committed transaction. Check its state before - // deciding whether the batch is complete or needs a new label. - checkLabelState(host, flushData.getLabel()); - return true; - } throw new StarRocksConnectorException( StarRocksConnectorErrorCode.FLUSH_DATA_FAILED, errorBuilder.toString()); -<<<<<<< HEAD - } else if (RESULT_LABEL_EXISTED.equals(loadResult.get(keyStatus))) { - LOG.debug("StreamLoad response:\n" + JsonUtils.toJsonString(loadResult)); - // has to block-checking the state to get the final result - checkLabelState(host, flushData.getLabel()); - return true; -======= ->>>>>>> 14a27ac26a0a71b56722f86cdf8ec2779094ceaf } // Publish Timeout, Label Already Exists, a Fail response blaming a reused label, or any // other non-terminal status: StarRocks received the request but the transaction outcome is @@ -293,21 +262,8 @@ private void checkLabelState(String host, String label, Map<String, Object> loadResult) throws IOException { int idx = 0; -<<<<<<< HEAD - while (true) { - try { - TimeUnit.SECONDS.sleep(Math.min(++idx, MAX_SLEEP_TIME)); - } catch (InterruptedException ex) { - Thread.currentThread().interrupt(); - throw new StarRocksConnectorException( - StarRocksConnectorErrorCode.FLUSH_DATA_FAILED, - String.format("Interrupted while checking the state of label[%s].", label), - ex); - } -======= long deadlineNanos = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(labelStateTimeoutMs); while (System.nanoTime() < deadlineNanos) { ->>>>>>> 14a27ac26a0a71b56722f86cdf8ec2779094ceaf try { String queryLoadStateUrl = new StringBuilder(host) --- a/seatunnel-connectors-v2/connector-starrocks/src/test/java/org/apache/seatunnel/connectors/seatunnel/starrocks/client/StarRocksSinkManagerTest.java +++ b/seatunnel-connectors-v2/connector-starrocks/src/test/java/org/apache/seatunnel/connectors/seatunnel/starrocks/client/StarRocksSinkManagerTest.java @@ -24,11 +24,8 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; -<<<<<<< HEAD -======= import java.util.ArrayList; import java.util.List; ->>>>>>> 14a27ac26a0a71b56722f86cdf8ec2779094ceaf import java.util.concurrent.atomic.AtomicInteger; import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; @@ -64,11 +61,7 @@ when(mockSinkConfig.getMaxRetries()).thenReturn(3); when(mockSinkConfig.getRetryBackoffMultiplierMs()).thenReturn(0); when(mockSinkConfig.getMaxRetryBackoffMs()).thenReturn(1000); -<<<<<<< HEAD - AtomicInteger labelSequence = new AtomicInteger(); -======= labelSequence = new AtomicInteger(); ->>>>>>> 14a27ac26a0a71b56722f86cdf8ec2779094ceaf this.sinkManager = new StarRocksSinkManager(mockSinkConfig, null, mockStreamLoadVisitor) { public String createBatchLabel() { @@ -83,47 +76,6 @@ * <p>A later successful flush must still send the record retained after retries are exhausted. */ @Test -<<<<<<< HEAD - void testUnclassifiedLabelMessageDoesNotChangeLabel() throws Exception { - AtomicInteger attempts = new AtomicInteger(); - when(mockStreamLoadVisitor.doStreamLoad(any())) - .thenAnswer( - invocation -> { - StarRocksFlushTuple tuple = invocation.getArgument(0); - if (attempts.incrementAndGet() == 1) { - assertEquals("test-label-1", tuple.getLabel()); - throw new RuntimeException( - "Label [test-label-1] has already been used"); - } - assertEquals("test-label-1", tuple.getLabel()); - return true; - }); - - sinkManager.write("test-record"); - - assertDoesNotThrow(() -> sinkManager.flush()); - verify(mockStreamLoadVisitor, times(2)).doStreamLoad(any()); - } - - @Test - void testLabelAlreadyUsedExhaustionPreservesBatchForNextFlush() throws Exception { - when(mockStreamLoadVisitor.doStreamLoad(any())) - .thenAnswer( - invocation -> { - StarRocksFlushTuple tuple = invocation.getArgument(0); - throw new RuntimeException( - "Label [" + tuple.getLabel() + "] has already been used"); - }); - - sinkManager.write("test-record"); - - assertThrows(StarRocksConnectorException.class, () -> sinkManager.flush()); - verify(mockStreamLoadVisitor, times(4)).doStreamLoad(any()); - - org.mockito.Mockito.reset(mockStreamLoadVisitor); - when(mockStreamLoadVisitor.doStreamLoad(any())).thenReturn(true); - -======= void testLabelAlreadyMessageDoesNotDiscardBufferedRecords() throws Exception { doThrow(new RuntimeException("Label [test-label-1] has already been used")) .when(mockStreamLoadVisitor) @@ -135,51 +87,9 @@ verify(mockStreamLoadVisitor, times(4)).doStreamLoad(any()); doReturn(true).when(mockStreamLoadVisitor).doStreamLoad(any()); ->>>>>>> 14a27ac26a0a71b56722f86cdf8ec2779094ceaf assertDoesNotThrow(() -> sinkManager.flush()); verify(mockStreamLoadVisitor, times(5)).doStreamLoad(any()); assertEquals(1, labelSequence.get()); - } - - @Test - void testLabelAlreadyUsedWithNoRetriesFailsWithoutDroppingBatch() throws Exception { - when(mockSinkConfig.getMaxRetries()).thenReturn(0); - when(mockStreamLoadVisitor.doStreamLoad(any())) - .thenAnswer( - invocation -> { - StarRocksFlushTuple tuple = invocation.getArgument(0); - throw new RuntimeException( - "Label [" + tuple.getLabel() + "] has already been used"); - }); - - sinkManager.write("test-record"); - - assertThrows(StarRocksConnectorException.class, () -> sinkManager.flush()); - verify(mockStreamLoadVisitor, times(1)).doStreamLoad(any()); - } - - @Test - void testReCreateLabelExceptionRetriesWithNewLabel() throws Exception { - AtomicInteger attempts = new AtomicInteger(); - when(mockStreamLoadVisitor.doStreamLoad(any())) - .thenAnswer( - invocation -> { - StarRocksFlushTuple tuple = invocation.getArgument(0); - if (attempts.incrementAndGet() == 1) { - assertEquals("test-label-1", tuple.getLabel()); - throw new StarRocksConnectorException( - StarRocksConnectorErrorCode.FLUSH_DATA_FAILED, - "The previous label cannot be reused", - true); - } - assertEquals("test-label-2", tuple.getLabel()); - return true; - }); - - sinkManager.write("test-record"); - - assertDoesNotThrow(() -> sinkManager.flush()); - verify(mockStreamLoadVisitor, times(2)).doStreamLoad(any()); } @Test --- a/seatunnel-connectors-v2/connector-starrocks/src/test/java/org/apache/seatunnel/connectors/seatunnel/starrocks/client/StarRocksStreamLoadVisitorTest.java +++ b/seatunnel-connectors-v2/connector-starrocks/src/test/java/org/apache/seatunnel/connectors/seatunnel/starrocks/client/StarRocksStreamLoadVisitorTest.java @@ -25,10 +25,7 @@ import java.io.IOException; import java.nio.charset.StandardCharsets; -<<<<<<< HEAD -======= import java.time.Duration; ->>>>>>> 14a27ac26a0a71b56722f86cdf8ec2779094ceaf import java.util.Collections; import java.util.HashMap; import java.util.List; @@ -37,19 +34,12 @@ import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertThrows; -<<<<<<< HEAD -import static org.junit.jupiter.api.Assertions.assertTrue; -import static org.mockito.ArgumentMatchers.any; -import static org.mockito.ArgumentMatchers.anyMap; -import static org.mockito.ArgumentMatchers.anyString; -======= import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyInt; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.atLeastOnce; ->>>>>>> 14a27ac26a0a71b56722f86cdf8ec2779094ceaf import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.times; @@ -136,25 +126,6 @@ }); } -<<<<<<< HEAD - @Test - void treatsVisibleReusedLabelAsSuccessful() throws Exception { - StarRocksStreamLoadVisitor visitor = visitorWithReusedLabelState("VISIBLE"); - - assertTrue(visitor.doStreamLoad(flushTuple())); - } - - @Test - void treatsCommittedReusedLabelAsSuccessful() throws Exception { - StarRocksStreamLoadVisitor visitor = visitorWithReusedLabelState("COMMITTED"); - - assertTrue(visitor.doStreamLoad(flushTuple())); - } - - @Test - void marksAbortedReusedLabelForRecreation() throws Exception { - StarRocksStreamLoadVisitor visitor = visitorWithReusedLabelState("ABORTED"); -======= /** * Verifies that Publish Timeout is resolved through the label state instead of being trusted as * success. @@ -225,21 +196,10 @@ when(httpHelper.doHttpGet(anyString(), any(), anyInt())).thenReturn(labelState); StarRocksStreamLoadVisitor visitor = new StarRocksStreamLoadVisitor(sinkConfig, createTableSchema(), httpHelper, 500); ->>>>>>> 14a27ac26a0a71b56722f86cdf8ec2779094ceaf StarRocksConnectorException exception = assertThrows( StarRocksConnectorException.class, -<<<<<<< HEAD - () -> visitor.doStreamLoad(flushTuple())); - - assertTrue(exception.needReCreateLabel()); - } - - @Test - void keepsUnknownReusedLabelFromBeingResubmitted() throws Exception { - StarRocksStreamLoadVisitor visitor = visitorWithReusedLabelState("UNKNOWN"); -======= () -> visitor.doStreamLoad(createFlushTuple())); assertFalse(exception.needReCreateLabel()); verify(httpHelper, atLeastOnce()).doHttpGet(anyString(), any(), anyInt()); @@ -379,54 +339,10 @@ when(httpHelper.doHttpGet(anyString(), any(), anyInt())).thenReturn(labelState); StarRocksStreamLoadVisitor visitor = new StarRocksStreamLoadVisitor(sinkConfig, createTableSchema(), httpHelper, 500); ->>>>>>> 14a27ac26a0a71b56722f86cdf8ec2779094ceaf StarRocksConnectorException exception = assertThrows( StarRocksConnectorException.class, -<<<<<<< HEAD - () -> visitor.doStreamLoad(flushTuple())); - - assertFalse(exception.needReCreateLabel()); - } - - @Test - void treatsLabelAlreadyExistsWithVisibleStateAsSuccessful() throws Exception { - StarRocksStreamLoadVisitor visitor = - visitorWithLoadResultAndLabelState("Label Already Exists", null, "VISIBLE"); - - assertTrue(visitor.doStreamLoad(flushTuple())); - } - - @Test - void interruptionWhileCheckingLabelStateFailsWithoutRecreatingLabel() throws Exception { - StarRocksStreamLoadVisitor visitor = visitorWithReusedLabelState("PREPARE"); - - Thread.currentThread().interrupt(); - try { - StarRocksConnectorException exception = - assertThrows( - StarRocksConnectorException.class, - () -> visitor.doStreamLoad(flushTuple())); - - assertFalse(exception.needReCreateLabel()); - assertTrue(Thread.currentThread().isInterrupted()); - } finally { - Thread.interrupted(); - } - } - - /** Builds a visitor whose stream-load response requires checking an existing label state. */ - private StarRocksStreamLoadVisitor visitorWithReusedLabelState(String labelState) - throws IOException { - return visitorWithLoadResultAndLabelState( - "Fail", "Label [test-label] has already been used", labelState); - } - - /** Builds a visitor with deterministic stream-load and label-state responses. */ - private StarRocksStreamLoadVisitor visitorWithLoadResultAndLabelState( - String status, String message, String labelState) throws IOException { -======= () -> visitor.doStreamLoad(createFlushTuple())); assertFalse(exception.needReCreateLabel()); verify(httpHelper, atLeastOnce()).doHttpGet(anyString(), any(), anyInt()); @@ -607,7 +523,6 @@ * @return mocked sink configuration for one JSON Stream Load target */ private SinkConfig createSinkConfig() { ->>>>>>> 14a27ac26a0a71b56722f86cdf8ec2779094ceaf SinkConfig sinkConfig = mock(SinkConfig.class); when(sinkConfig.getLoadFormat()).thenReturn(SinkConfig.StreamLoadFormat.JSON); when(sinkConfig.getBatchMaxBytes()).thenReturn(1024L); @@ -615,33 +530,6 @@ when(sinkConfig.getNodeUrls()).thenReturn(Collections.singletonList("localhost:8030")); when(sinkConfig.getDatabase()).thenReturn("test_db"); when(sinkConfig.getTable()).thenReturn("test_table"); -<<<<<<< HEAD - when(sinkConfig.getUsername()).thenReturn("user"); - when(sinkConfig.getPassword()).thenReturn("password"); - - TableSchema tableSchema = mock(TableSchema.class); - when(tableSchema.getColumns()).thenReturn(Collections.emptyList()); - - HttpHelper httpHelper = mock(HttpHelper.class); - when(httpHelper.tryHttpConnection("http://localhost:8030";)).thenReturn(true); - Map<String, Object> loadResult = new HashMap<>(); - loadResult.put("Status", status); - if (message != null) { - loadResult.put("Message", message); - } - when(httpHelper.doHttpPut(anyString(), any(byte[].class), anyMap())).thenReturn(loadResult); - when(httpHelper.doHttpGet(anyString(), anyMap())) - .thenReturn(Collections.singletonMap("state", labelState)); - - return new StarRocksStreamLoadVisitor(sinkConfig, tableSchema, httpHelper); - } - - /** Creates one JSON row under the label used by the response-state tests. */ - private StarRocksFlushTuple flushTuple() { - byte[] row = "{}".getBytes(StandardCharsets.UTF_8); - return new StarRocksFlushTuple( - "test-label", (long) row.length, Collections.singletonList(row)); -======= when(sinkConfig.getUsername()).thenReturn("test_user"); when(sinkConfig.getPassword()).thenReturn("test_password"); when(sinkConfig.getStreamLoadProps()).thenReturn(Collections.emptyMap()); @@ -678,6 +566,5 @@ Map<String, Object> loadResult = new HashMap<>(); loadResult.put("Status", status); return loadResult; ->>>>>>> 14a27ac26a0a71b56722f86cdf8ec2779094ceaf } } ``` </details> <!-- contributor-patch:10680:0570563e5bf5892908e0776396ed70d09bcfba3c --> -- 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]
