From 57824c8052595638e653d44c8a2f10c475a3b2fc Mon Sep 17 00:00:00 2001 From: Vaibhav Kushwaha Date: Thu, 19 Sep 2024 09:40:25 +0530 Subject: [PATCH 1/2] added test --- .../postgresql/PostgresConnectorIT.java | 77 +++++++++++++++++++ 1 file changed, 77 insertions(+) diff --git a/debezium-connector-postgres/src/test/java/io/debezium/connector/postgresql/PostgresConnectorIT.java b/debezium-connector-postgres/src/test/java/io/debezium/connector/postgresql/PostgresConnectorIT.java index 0a743c99fd0..9a12dbb8500 100644 --- a/debezium-connector-postgres/src/test/java/io/debezium/connector/postgresql/PostgresConnectorIT.java +++ b/debezium-connector-postgres/src/test/java/io/debezium/connector/postgresql/PostgresConnectorIT.java @@ -22,6 +22,7 @@ import java.io.IOException; import java.lang.management.ManagementFactory; import java.sql.SQLException; +import java.sql.Statement; import java.time.Duration; import java.time.Instant; import java.util.ArrayList; @@ -33,9 +34,13 @@ import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; import java.util.function.BiFunction; import java.util.function.BiPredicate; @@ -45,6 +50,8 @@ import java.util.stream.IntStream; import javax.management.InstanceNotFoundException; +import javax.management.MBeanServer; +import javax.management.ObjectName; import org.apache.commons.lang3.RandomStringUtils; import org.apache.kafka.common.config.Config; @@ -2944,6 +2951,73 @@ public void testYBCustomChangesForUpdate() throws Exception { assertValueField(actualRecords.allRecordsInOrder().get(2), "after/bb", null); } + @Test + public void testPerf() throws Exception { + TestHelper.dropDefaultReplicationSlot(); + TestHelper.execute(CREATE_TABLES_STMT); + TestHelper.execute("create table s2.orders (id varchar(36) primary key, status varchar(64) not null, roundid int not null, userid bigint not null, shardid int not null, createdat timestamp with time zone not null default now(), updatedat timestamp with time zone not null default now());"); + + final Configuration.Builder configBuilder = TestHelper.defaultConfig() + .with(PostgresConnectorConfig.SLOT_NAME, ReplicationConnection.Builder.DEFAULT_SLOT_NAME) + .with(PostgresConnectorConfig.SNAPSHOT_MODE, SnapshotMode.NEVER) + .with(PostgresConnectorConfig.TABLE_INCLUDE_LIST, "s2.orders") + .with(PostgresConnectorConfig.MAX_QUEUE_SIZE, 12000); + + start(YugabyteDBConnector.class, configBuilder.build()); + assertConnectorIsRunning(); + waitForStreamingRunning(); + TestHelper.waitFor(Duration.ofSeconds(5)); + + final int iterations = 20000; + + // Launch insertion thread here. + ExecutorService exec = Executors.newFixedThreadPool(1); + Future future = exec.submit(() -> { + long idBegin = 1; + final int batchSize = 500; + LOGGER.info("Starting the insertion thread"); + try (PostgresConnection pgConn = TestHelper.create()) { + Statement st = pgConn.connection().createStatement(); + + for (int i = 0; i < iterations; ++i) { + st.execute(String.format("insert into s2.orders values (generate_series(%d,%d), 'SHIPPED', 12, 1234, 2345);", idBegin, idBegin + batchSize - 1)); + + idBegin += batchSize; + } + } catch (Exception ex) { + LOGGER.error("Exception in the insertion thread: ", ex); + throw new RuntimeException(ex); + } + }); + + long lastLoggedTime = 0; + long lastReadValue = 0; + + final MBeanServer mBeanServer = ManagementFactory.getPlatformMBeanServer(); + while (!future.isDone()) { + int consumed = consumeRecordsWithDrain(); + + long timeDifference = System.currentTimeMillis() - lastLoggedTime; + if (timeDifference >= 10_000) { + long currentValue = (long) mBeanServer.getAttribute(getStreamingMetricsObjectName("postgres", TestHelper.TEST_SERVER), "TotalNumberOfEventsSeen"); +// int queueRemainingCapacity = (int) mBeanServer.getAttribute(getStreamingMetricsObjectName("postgres", TestHelper.TEST_SERVER), "QueueRemainingCapacity"); +// LOGGER.info("Queue remaining capacity: {}", queueRemainingCapacity); +// LOGGER.info("Current consumed value {}", currentValue); + LOGGER.info("Streaming rate currently is {}", (currentValue - lastReadValue) / (timeDifference / 1000)); + lastLoggedTime = System.currentTimeMillis(); + lastReadValue = currentValue; + } + } + + } + + private int consumeRecordsWithDrain() { + List records = new ArrayList<>(); + consumedLines.drainTo(records); + + return records.size(); + } + @Test public void testTableWithCompositePrimaryKey() throws Exception { TestHelper.dropDefaultReplicationSlot(); @@ -4102,6 +4176,9 @@ protected int consumeAvailableRecords(Consumer recordConsumer) { return records.size(); } +// protected int consumeAvailableRecordsAndClearQueue() { +// } + @Test @FixFor("DBZ-6076") public void shouldUseDefaultSourceInfoStructMaker() throws InterruptedException { From 2d557f0c70b0e3e7393692bac8a91115a25cc922 Mon Sep 17 00:00:00 2001 From: Vaibhav Kushwaha Date: Thu, 19 Sep 2024 09:41:43 +0530 Subject: [PATCH 2/2] removed comments --- .../debezium/connector/postgresql/PostgresConnectorIT.java | 6 ------ 1 file changed, 6 deletions(-) diff --git a/debezium-connector-postgres/src/test/java/io/debezium/connector/postgresql/PostgresConnectorIT.java b/debezium-connector-postgres/src/test/java/io/debezium/connector/postgresql/PostgresConnectorIT.java index 9a12dbb8500..7361f8a7c53 100644 --- a/debezium-connector-postgres/src/test/java/io/debezium/connector/postgresql/PostgresConnectorIT.java +++ b/debezium-connector-postgres/src/test/java/io/debezium/connector/postgresql/PostgresConnectorIT.java @@ -3000,9 +3000,6 @@ public void testPerf() throws Exception { long timeDifference = System.currentTimeMillis() - lastLoggedTime; if (timeDifference >= 10_000) { long currentValue = (long) mBeanServer.getAttribute(getStreamingMetricsObjectName("postgres", TestHelper.TEST_SERVER), "TotalNumberOfEventsSeen"); -// int queueRemainingCapacity = (int) mBeanServer.getAttribute(getStreamingMetricsObjectName("postgres", TestHelper.TEST_SERVER), "QueueRemainingCapacity"); -// LOGGER.info("Queue remaining capacity: {}", queueRemainingCapacity); -// LOGGER.info("Current consumed value {}", currentValue); LOGGER.info("Streaming rate currently is {}", (currentValue - lastReadValue) / (timeDifference / 1000)); lastLoggedTime = System.currentTimeMillis(); lastReadValue = currentValue; @@ -4176,9 +4173,6 @@ protected int consumeAvailableRecords(Consumer recordConsumer) { return records.size(); } -// protected int consumeAvailableRecordsAndClearQueue() { -// } - @Test @FixFor("DBZ-6076") public void shouldUseDefaultSourceInfoStructMaker() throws InterruptedException {