SpringCloud/spring-cloud-stream-samples
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Adding scheam registry samples

Browse Source 
* Add plain vanilla schema registry sammples using ScSt schema registry server
 * Add samples using Confluent schema registry
 * Add acceptance tests for the plain vanilla examples
This commit is contained in:
Soby Chacko
2018-03-17 14:04:08 -04:00
parent 7c3b457f62
commit 7569127dea
69 changed files with 5012 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

51
samples-acceptance-tests/src/test/java/sample/acceptance/tests/AbstractSampleTests.java Normal file
View File

@@ -0,0 +1,51 @@
package sample.acceptance.tests;
import org.assertj.core.util.Files;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.StringUtils;
import java.io.File;
import java.nio.charset.StandardCharsets;
import java.util.stream.Stream;
import static org.junit.Assert.fail;
/**
* @author Soby Chacko
*/
public abstract class AbstractSampleTests {
private static final Logger logger = LoggerFactory.getLogger(AbstractSampleTests.class);
protected boolean waitForLogEntryInFile(String app, File f, String... entries) {
logger.info("Looking for '" + StringUtils.arrayToCommaDelimitedString(entries) + "' in logfile for " + app);
long timeout = System.currentTimeMillis() + (60 * 1000);
boolean exists = false;
while (!exists && System.currentTimeMillis() < timeout) {
try {
Thread.sleep(2 * 1000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new IllegalStateException(e.getMessage(), e);
}
logger.info("Polling to get log file. Remaining poll time = "
+ (timeout - System.currentTimeMillis() + " ms."));
String log = Files.contentOf(f, StandardCharsets.UTF_8);
if (log != null) {
if (Stream.of(entries).allMatch(log::contains)) {
exists = true;
}
}
}
if (exists) {
logger.info("Matched all '" + StringUtils.arrayToCommaDelimitedString(entries) + "' in logfile for app " + app);
} else {
logger.error("ERROR: Couldn't find all '" + StringUtils.arrayToCommaDelimitedString(entries) + "' in logfile for " + app);
fail("Could not find the test data in the logs");
}
return true;
}
}

67
samples-acceptance-tests/src/test/java/sample/acceptance/tests/SampleAcceptanceTests.java
View File

@@ -23,13 +23,10 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.datasource.SingleConnectionDataSource;
import org.springframework.util.StringUtils;
import org.springframework.web.client.RestTemplate;
import javax.sql.DataSource;
import java.io.File;
import java.nio.charset.StandardCharsets;
import java.util.stream.Stream;
import static org.junit.Assert.fail;
@@ -41,7 +38,7 @@ import static org.junit.Assert.fail;
*
* @author Soby Chacko
*/
public class SampleAcceptanceTests {
public class SampleAcceptanceTests extends AbstractSampleTests {
private static final Logger logger = LoggerFactory.getLogger(SampleAcceptanceTests.class);
@@ -108,34 +105,6 @@ public class SampleAcceptanceTests {
verifyJdbcSink();
}
private void verifyJdbcSink() {
JdbcTemplate db;
DataSource dataSource = new SingleConnectionDataSource("jdbc:mariadb://localhost:3306/sample_mysql_db",
"root", "pwd", false);
db = new JdbcTemplate(dataSource);
long timeout = System.currentTimeMillis() + (30 * 1000);
boolean exists = false;
while (!exists && System.currentTimeMillis() < timeout) {
try {
Thread.sleep(5 * 1000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new IllegalStateException(e.getMessage(), e);
}
Integer count = db.queryForObject("select count(*) from test", Integer.class);
if (count > 0) {
exists = true;
}
}
if (!exists) {
fail("No records found in database!");
}
}
@Test
public void testDynamicSourceSampleKafka() throws Exception {
ProcessBuilder pb = new ProcessBuilder("java", "-jar", "/tmp/dynamic-destination-source-kafka-sample.jar", "--management.endpoints.web.exposure.include=*");
@@ -345,35 +314,31 @@ public class SampleAcceptanceTests {
Files.delete(file);
}
boolean waitForLogEntryInFile(String app, File f, String... entries) {
logger.info("Looking for '" + StringUtils.arrayToCommaDelimitedString(entries) + "' in logfile for " + app);
long timeout = System.currentTimeMillis() + (60 * 1000);
private void verifyJdbcSink() {
JdbcTemplate db;
DataSource dataSource = new SingleConnectionDataSource("jdbc:mariadb://localhost:3306/sample_mysql_db",
"root", "pwd", false);
db = new JdbcTemplate(dataSource);
long timeout = System.currentTimeMillis() + (30 * 1000);
boolean exists = false;
while (!exists && System.currentTimeMillis() < timeout) {
try {
Thread.sleep(2 * 1000);
Thread.sleep(5 * 1000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new IllegalStateException(e.getMessage(), e);
}
if (!exists) {
logger.info("Polling to get log file. Remaining poll time = "
+ (timeout - System.currentTimeMillis() + " ms."));
String log = Files.contentOf(f, StandardCharsets.UTF_8);
if (log != null) {
if (Stream.of(entries).allMatch(s -> log.contains(s))) {
exists = true;
}
}
Integer count = db.queryForObject("select count(*) from test", Integer.class);
if (count > 0) {
exists = true;
}
}
if (exists) {
logger.info("Matched all '" + StringUtils.arrayToCommaDelimitedString(entries) + "' in logfile for app " + app);
} else {
logger.error("ERROR: Couldn't find all '" + StringUtils.arrayToCommaDelimitedString(entries) + "' in logfile for " + app);
if (!exists) {
fail("No records found in database!");
}
return exists;
}
}

117
samples-acceptance-tests/src/test/java/sample/acceptance/tests/SchemaRegistryVanillaSampleTests.java Normal file
View File

@@ -0,0 +1,117 @@
package sample.acceptance.tests;
import org.assertj.core.util.Files;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
import org.springframework.web.client.RestTemplate;
import java.io.File;
import static org.junit.Assert.fail;
/**
* @author Soby Chacko
*/
public class SchemaRegistryVanillaSampleTests extends AbstractSampleTests {
private static final Logger logger = LoggerFactory.getLogger(SchemaRegistryVanillaSampleTests.class);
@Test
public void testSchemaRegistryVanillaKafka() throws Exception {
runAgainstMiddleware("/tmp/schema-registry-vanilla-registry-kafka.jar",
"/tmp/schema-registry-vanilla-consumer-kafka.jar",
"/tmp/schema-registry-vanilla-producer1-kafka.jar",
"/tmp/schema-registry-vanilla-producer2-kafka.jar");
}
@Test
public void testSchemaRegistryVanillaRabbit() throws Exception {
runAgainstMiddleware("/tmp/schema-registry-vanilla-registry-rabbit.jar",
"/tmp/schema-registry-vanilla-consumer-rabbit.jar",
"/tmp/schema-registry-vanilla-producer1-rabbit.jar",
"/tmp/schema-registry-vanilla-producer2-rabbit.jar");
}
private void runAgainstMiddleware(String registryJar, String consumerJar, String producer1Jar, String producer2Jar) throws Exception {
Process registryProcess = null;
Process consumerProcess = null;
Process producer1Process = null;
Process producer2Process = null;
try {
ProcessBuilder pbRegistry = new ProcessBuilder("java", "-jar", registryJar);
File registryFile = Files.newTemporaryFile();
logger.info("Output is redirected to " + registryFile.getAbsolutePath());
pbRegistry.redirectOutput(registryFile);
registryProcess = pbRegistry.start();
waitForLogEntryInFile("Schema Registry Vanilla Server", registryFile, "Started RegistryApplication in");
ProcessBuilder pbConsumer = new ProcessBuilder("java", "-jar", consumerJar);
File consumerFile = Files.newTemporaryFile();
logger.info("Output is redirected to " + consumerFile.getAbsolutePath());
pbConsumer.redirectOutput(consumerFile);
consumerProcess = pbConsumer.start();
waitForLogEntryInFile("Schema Registry Vanilla Consumer", consumerFile, "Started ConsumerApplication in");
ProcessBuilder pbProducer1 = new ProcessBuilder("java", "-jar", producer1Jar);
File producer1File = Files.newTemporaryFile();
logger.info("Output is redirected to " + producer1File.getAbsolutePath());
pbProducer1.redirectOutput(producer1File);
producer1Process = pbProducer1.start();
waitForLogEntryInFile("Schema Registry Vanilla Producer1", producer1File, "Started Producer1Application in");
ProcessBuilder pbProducer2 = new ProcessBuilder("java", "-jar", producer2Jar);
File producer2File = Files.newTemporaryFile();
logger.info("Output is redirected to " + producer2File.getAbsolutePath());
pbProducer2.redirectOutput(producer2File);
producer2Process = pbProducer2.start();
waitForLogEntryInFile("Schema Registry Vanilla Producer2", producer2File, "Started Producer2Application in");
RestTemplate restTemplate = new RestTemplate();
MultiValueMap<String, Object> parametersMap = new LinkedMultiValueMap<>();
parametersMap.add("id", "foobar");
parametersMap.add("temperature", 30);
parametersMap.add("acceleration", 10);
parametersMap.add("velocity", 20);
restTemplate.postForObject(
"http://localhost:9009/messagesX", parametersMap, String.class);
boolean found = waitForLogEntryInFile("Schema Registry Vanilla Consumer", consumerFile,
"{\"id\": \"foobar-v1\", \"internalTemperature\": 30.0, \"externalTemperature\": 0.0, \"acceleration\": 10.0, \"velocity\": 20.0}");
if (!found) {
fail("Could not find the test data in the logs");
}
restTemplate.postForObject(
"http://localhost:9010/messagesX", parametersMap, String.class);
waitForLogEntryInFile("Schema Registry Vanilla Consumer", consumerFile,
"{\"id\": \"foobar-v2\", \"internalTemperature\": 30.0, \"externalTemperature\": 0.0, \"acceleration\": 10.0, \"velocity\": 20.0}");
} finally {
if (registryProcess != null) {
registryProcess.destroyForcibly();
}
if (consumerProcess != null) {
consumerProcess.destroyForcibly();
}
if (producer1Process != null) {
producer1Process.destroyForcibly();
}
if (producer2Process != null) {
producer2Process.destroyForcibly();
}
}
}
}