|
| 1 | +/* |
| 2 | + * Licensed to the Apache Software Foundation (ASF) under one or more |
| 3 | + * contributor license agreements. See the NOTICE file distributed with |
| 4 | + * this work for additional information regarding copyright ownership. |
| 5 | + * The ASF licenses this file to You under the Apache License, Version 2.0 |
| 6 | + * (the "License"); you may not use this file except in compliance with |
| 7 | + * the License. You may obtain a copy of the License at |
| 8 | + * |
| 9 | + * http://www.apache.org/licenses/LICENSE-2.0 |
| 10 | + * |
| 11 | + * Unless required by applicable law or agreed to in writing, software |
| 12 | + * distributed under the License is distributed on an "AS IS" BASIS, |
| 13 | + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 14 | + * See the License for the specific language governing permissions and |
| 15 | + * limitations under the License. |
| 16 | + */ |
| 17 | + |
| 18 | +package org.apache.seatunnel.e2e.connector.doris; |
| 19 | + |
| 20 | +import org.apache.seatunnel.e2e.common.container.ContainerExtendedFactory; |
| 21 | +import org.apache.seatunnel.e2e.common.container.EngineType; |
| 22 | +import org.apache.seatunnel.e2e.common.container.TestContainer; |
| 23 | +import org.apache.seatunnel.e2e.common.junit.DisabledOnContainer; |
| 24 | +import org.apache.seatunnel.e2e.common.junit.TestContainerExtension; |
| 25 | + |
| 26 | +import org.junit.jupiter.api.Assertions; |
| 27 | +import org.junit.jupiter.api.TestTemplate; |
| 28 | +import org.testcontainers.containers.Container; |
| 29 | + |
| 30 | +import lombok.extern.slf4j.Slf4j; |
| 31 | + |
| 32 | +import java.io.IOException; |
| 33 | +import java.sql.SQLException; |
| 34 | +import java.sql.Statement; |
| 35 | +import java.util.concurrent.CompletableFuture; |
| 36 | +import java.util.concurrent.ExecutionException; |
| 37 | +import java.util.concurrent.TimeUnit; |
| 38 | + |
| 39 | +import static org.awaitility.Awaitility.given; |
| 40 | + |
| 41 | +@Slf4j |
| 42 | +public class DorisErrorIT extends AbstractDorisIT { |
| 43 | + private static final String TABLE = "doris_e2e_table"; |
| 44 | + private static final String DRIVER_JAR = |
| 45 | + "https://repo1.maven.org/maven2/com/mysql/mysql-connector-j/8.0.32/mysql-connector-j-8.0.32.jar"; |
| 46 | + |
| 47 | + private static final String sinkDB = "e2e_sink"; |
| 48 | + |
| 49 | + @TestContainerExtension |
| 50 | + protected final ContainerExtendedFactory extendedFactory = |
| 51 | + container -> { |
| 52 | + Container.ExecResult extraCommands = |
| 53 | + container.execInContainer( |
| 54 | + "bash", |
| 55 | + "-c", |
| 56 | + "mkdir -p /tmp/seatunnel/plugins/jdbc/lib && cd /tmp/seatunnel/plugins/jdbc/lib && wget " |
| 57 | + + DRIVER_JAR); |
| 58 | + Assertions.assertEquals(0, extraCommands.getExitCode(), extraCommands.getStderr()); |
| 59 | + }; |
| 60 | + |
| 61 | + @TestTemplate |
| 62 | + @DisabledOnContainer( |
| 63 | + value = {}, |
| 64 | + type = {EngineType.SPARK, EngineType.FLINK}, |
| 65 | + disabledReason = "flink/spark failed reason not same") |
| 66 | + public void testDoris(TestContainer container) throws InterruptedException, ExecutionException { |
| 67 | + initializeJdbcTable(); |
| 68 | + CompletableFuture<Container.ExecResult> future = |
| 69 | + CompletableFuture.supplyAsync( |
| 70 | + () -> { |
| 71 | + try { |
| 72 | + return container.executeJob( |
| 73 | + "/fake_source_and_doris_sink_timeout_error.conf"); |
| 74 | + } catch (IOException | InterruptedException e) { |
| 75 | + throw new RuntimeException(e); |
| 76 | + } |
| 77 | + }); |
| 78 | + // wait for the job to start |
| 79 | + Thread.sleep(10 * 1000); |
| 80 | + super.container.stop(); |
| 81 | + Assertions.assertNotEquals(0, future.get().getExitCode()); |
| 82 | + super.container.start(); |
| 83 | + // wait for the container to restart |
| 84 | + given().ignoreExceptions() |
| 85 | + .await() |
| 86 | + .atMost(10000, TimeUnit.SECONDS) |
| 87 | + .untilAsserted(this::initializeJdbcConnection); |
| 88 | + } |
| 89 | + |
| 90 | + private void initializeJdbcTable() { |
| 91 | + try { |
| 92 | + try (Statement statement = jdbcConnection.createStatement()) { |
| 93 | + // create test databases |
| 94 | + statement.execute(createDatabase(sinkDB)); |
| 95 | + log.info("create sink database succeed"); |
| 96 | + // create sink table |
| 97 | + statement.execute(createTableForTest(sinkDB)); |
| 98 | + } catch (SQLException e) { |
| 99 | + throw new RuntimeException("Initializing table failed!", e); |
| 100 | + } |
| 101 | + } catch (Exception e) { |
| 102 | + throw new RuntimeException("Initializing jdbc failed!", e); |
| 103 | + } |
| 104 | + } |
| 105 | + |
| 106 | + private String createDatabase(String db) { |
| 107 | + return String.format("CREATE DATABASE IF NOT EXISTS %s ;", db); |
| 108 | + } |
| 109 | + |
| 110 | + private String createTableForTest(String db) { |
| 111 | + String createTableSql = |
| 112 | + "create table if not exists `%s`.`%s`(\n" |
| 113 | + + "F_ID bigint null,\n" |
| 114 | + + "F_INT int null,\n" |
| 115 | + + "F_BIGINT bigint null,\n" |
| 116 | + + "F_TINYINT tinyint null,\n" |
| 117 | + + "F_SMALLINT smallint null,\n" |
| 118 | + + "F_DECIMAL decimal(18,6) null,\n" |
| 119 | + + "F_LARGEINT largeint null,\n" |
| 120 | + + "F_BOOLEAN boolean null,\n" |
| 121 | + + "F_DOUBLE double null,\n" |
| 122 | + + "F_FLOAT float null,\n" |
| 123 | + + "F_CHAR char null,\n" |
| 124 | + + "F_VARCHAR_11 varchar(11) null,\n" |
| 125 | + + "F_STRING string null,\n" |
| 126 | + + "F_DATETIME_P datetime(6),\n" |
| 127 | + + "F_DATETIME datetime,\n" |
| 128 | + + "F_DATE date\n" |
| 129 | + + ")\n" |
| 130 | + + "duplicate KEY(`F_ID`)\n" |
| 131 | + + "DISTRIBUTED BY HASH(`F_ID`) BUCKETS 1\n" |
| 132 | + + "properties(\n" |
| 133 | + + "\"replication_allocation\" = \"tag.location.default: 1\"" |
| 134 | + + ");"; |
| 135 | + return String.format(createTableSql, db, TABLE); |
| 136 | + } |
| 137 | +} |
0 commit comments