This is an automated email from the ASF dual-hosted git repository.
mmerli pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 7861a78 Change logging for debezium key value pairs (#5172)
7861a78 is described below
commit 7861a78b9fdbb17275b0e4fd910d6a150f88496b
Author: Ali Ahmed <alahmed...@gmail.com>
AuthorDate: Thu Sep 12 09:00:09 2019 -0700
Change logging for debezium key value pairs (#5172)
* Make naming consistent for Debezium postgreSql
* Change logging for debezium key value pairs
---
...ainer.java => DebeziumPostgreSqlContainer.java} | 4 ++--
.../integration/functions/PulsarFunctionsTest.java | 14 +++++++-------
.../integration/io/DebeziumMySqlSourceTester.java | 6 +++---
...er.java => DebeziumPostgreSqlSourceTester.java} | 22 +++++++++++-----------
4 files changed, 23 insertions(+), 23 deletions(-)
diff --git
a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/DebeziumPostgresqlContainer.java
b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/DebeziumPostgreSqlContainer.java
similarity index 91%
rename from
tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/DebeziumPostgresqlContainer.java
rename to
tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/DebeziumPostgreSqlContainer.java
index e85c27c..8f32c05 100644
---
a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/DebeziumPostgresqlContainer.java
+++
b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/containers/DebeziumPostgreSqlContainer.java
@@ -21,14 +21,14 @@ package org.apache.pulsar.tests.integration.containers;
import org.testcontainers.containers.wait.strategy.HostPortWaitStrategy;
-public class DebeziumPostgresqlContainer extends
ChaosContainer<DebeziumPostgresqlContainer> {
+public class DebeziumPostgreSqlContainer extends
ChaosContainer<DebeziumPostgreSqlContainer> {
public static final String NAME = "debezium-postgresql-example";
static final Integer[] PORTS = { 5432 };
private static final String IMAGE_NAME = "debezium/example-postgres:0.10";
- public DebeziumPostgresqlContainer(String clusterName) {
+ public DebeziumPostgreSqlContainer(String clusterName) {
super(clusterName, IMAGE_NAME);
this.withEnv("POSTGRES_USER", "postgres");
this.withEnv("POSTGRES_PASSWORD", "postgres");
diff --git
a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java
b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java
index 7fda45b..23beec8 100644
---
a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java
+++
b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/functions/PulsarFunctionsTest.java
@@ -45,7 +45,7 @@ import org.apache.pulsar.common.schema.KeyValue;
import org.apache.pulsar.functions.api.examples.AutoSchemaFunction;
import org.apache.pulsar.functions.api.examples.serde.CustomObject;
import org.apache.pulsar.tests.integration.containers.DebeziumMySQLContainer;
-import
org.apache.pulsar.tests.integration.containers.DebeziumPostgresqlContainer;
+import
org.apache.pulsar.tests.integration.containers.DebeziumPostgreSqlContainer;
import org.apache.pulsar.tests.integration.docker.ContainerExecException;
import org.apache.pulsar.tests.integration.docker.ContainerExecResult;
import org.apache.pulsar.tests.integration.functions.utils.CommandGenerator;
@@ -119,8 +119,8 @@ public abstract class PulsarFunctionsTest extends
PulsarFunctionsTestBase {
}
@Test
- public void testDebeziumPostgresqlSource() throws Exception {
- testDebeziumPostgresqlConnect();
+ public void testDebeziumPostgreSqlSource() throws Exception {
+ testDebeziumPostgreSqlConnect();
}
private void testSink(SinkTester tester, boolean builtin) throws Exception
{
@@ -1966,7 +1966,7 @@ public abstract class PulsarFunctionsTest extends
PulsarFunctionsTestBase {
getSourceInfoNotFound(tenant, namespace, sourceName);
}
- private void testDebeziumPostgresqlConnect() throws Exception {
+ private void testDebeziumPostgreSqlConnect() throws Exception {
final String tenant = TopicName.PUBLIC_TENANT;
final String namespace = TopicName.DEFAULT_NAMESPACE;
@@ -1991,11 +1991,11 @@ public abstract class PulsarFunctionsTest extends
PulsarFunctionsTestBase {
.subscribe();
@Cleanup
- DebeziumPostgresqlSourceTester sourceTester = new
DebeziumPostgresqlSourceTester(pulsarCluster);
+ DebeziumPostgreSqlSourceTester sourceTester = new
DebeziumPostgreSqlSourceTester(pulsarCluster);
// setup debezium postgresql server
- DebeziumPostgresqlContainer postgresqlContainer = new
DebeziumPostgresqlContainer(pulsarCluster.getClusterName());
- sourceTester.setServiceContainer(postgresqlContainer);
+ DebeziumPostgreSqlContainer postgreSqlContainer = new
DebeziumPostgreSqlContainer(pulsarCluster.getClusterName());
+ sourceTester.setServiceContainer(postgreSqlContainer);
// prepare the testing environment for source
prepareSource(sourceTester);
diff --git
a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/DebeziumMySqlSourceTester.java
b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/DebeziumMySqlSourceTester.java
index 2d1b4b5..580bf2c 100644
---
a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/DebeziumMySqlSourceTester.java
+++
b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/DebeziumMySqlSourceTester.java
@@ -91,9 +91,9 @@ public class DebeziumMySqlSourceTester extends
SourceTester<DebeziumMySQLContain
Message<KeyValue<byte[], byte[]>> msg = consumer.receive(2,
TimeUnit.SECONDS);
while(msg != null) {
recordsNumber ++;
- log.info("Received message: {}.", msg.getValue());
- String key = new String(msg.getValue().getKey());
- String value = new String(msg.getValue().getValue());
+ final String key = new String(msg.getValue().getKey());
+ final String value = new String(msg.getValue().getValue());
+ log.info("Received message: key = {}, value = {}.", key, value);
Assert.assertTrue(key.contains("dbserver1.inventory.products.Key"));
Assert.assertTrue(value.contains("dbserver1.inventory.products.Value"));
consumer.acknowledge(msg);
diff --git
a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/DebeziumPostgresqlSourceTester.java
b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/DebeziumPostgreSqlSourceTester.java
similarity index 84%
rename from
tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/DebeziumPostgresqlSourceTester.java
rename to
tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/DebeziumPostgreSqlSourceTester.java
index e95db22..d246258 100644
---
a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/DebeziumPostgresqlSourceTester.java
+++
b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/io/DebeziumPostgreSqlSourceTester.java
@@ -23,7 +23,7 @@ import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.common.schema.KeyValue;
-import
org.apache.pulsar.tests.integration.containers.DebeziumPostgresqlContainer;
+import
org.apache.pulsar.tests.integration.containers.DebeziumPostgreSqlContainer;
import org.apache.pulsar.tests.integration.containers.PulsarContainer;
import org.apache.pulsar.tests.integration.topologies.PulsarCluster;
import org.testng.Assert;
@@ -42,23 +42,23 @@ import java.util.concurrent.TimeUnit;
* which is a Postgresql database server preconfigured with an inventory
database.
*/
@Slf4j
-public class DebeziumPostgresqlSourceTester extends
SourceTester<DebeziumPostgresqlContainer> implements Closeable {
+public class DebeziumPostgreSqlSourceTester extends
SourceTester<DebeziumPostgreSqlContainer> implements Closeable {
private static final String NAME = "debezium-postgres";
private final String pulsarServiceUrl;
@Getter
- private DebeziumPostgresqlContainer debeziumPostgresqlContainer;
+ private DebeziumPostgreSqlContainer debeziumPostgresqlContainer;
private final PulsarCluster pulsarCluster;
- public DebeziumPostgresqlSourceTester(PulsarCluster cluster) {
+ public DebeziumPostgreSqlSourceTester(PulsarCluster cluster) {
super(NAME);
this.pulsarCluster = cluster;
pulsarServiceUrl = "pulsar://pulsar-proxy:" +
PulsarContainer.BROKER_PORT;
- sourceConfig.put("database.hostname",
DebeziumPostgresqlContainer.NAME);
+ sourceConfig.put("database.hostname",
DebeziumPostgreSqlContainer.NAME);
sourceConfig.put("database.port", "5432");
sourceConfig.put("database.user", "postgres");
sourceConfig.put("database.password", "postgres");
@@ -71,10 +71,10 @@ public class DebeziumPostgresqlSourceTester extends
SourceTester<DebeziumPostgre
}
@Override
- public void setServiceContainer(DebeziumPostgresqlContainer container) {
+ public void setServiceContainer(DebeziumPostgreSqlContainer container) {
log.info("start debezium postgresql server container.");
debeziumPostgresqlContainer = container;
- pulsarCluster.startService(DebeziumPostgresqlContainer.NAME,
debeziumPostgresqlContainer);
+ pulsarCluster.startService(DebeziumPostgreSqlContainer.NAME,
debeziumPostgresqlContainer);
}
@Override
@@ -93,9 +93,9 @@ public class DebeziumPostgresqlSourceTester extends
SourceTester<DebeziumPostgre
Message<KeyValue<byte[], byte[]>> msg = consumer.receive(2,
TimeUnit.SECONDS);
while(msg != null) {
recordsNumber ++;
- log.info("Received message: {}.", msg.getValue());
- String key = new String(msg.getValue().getKey());
- String value = new String(msg.getValue().getValue());
+ final String key = new String(msg.getValue().getKey());
+ final String value = new String(msg.getValue().getValue());
+ log.info("Received message: key = {}, value = {}.", key, value);
Assert.assertTrue(key.contains("dbserver1.inventory.products.Key"));
Assert.assertTrue(value.contains("dbserver1.inventory.products.Value"));
consumer.acknowledge(msg);
@@ -109,7 +109,7 @@ public class DebeziumPostgresqlSourceTester extends
SourceTester<DebeziumPostgre
@Override
public void close() {
if (pulsarCluster != null) {
- pulsarCluster.stopService(DebeziumPostgresqlContainer.NAME,
debeziumPostgresqlContainer);
+ pulsarCluster.stopService(DebeziumPostgreSqlContainer.NAME,
debeziumPostgresqlContainer);
}
}
