{"type":"doc","content":[{"type":"extension","attrs":{"layout":"default","extensionType":"com.atlassian.confluence.macro.core","extensionKey":"toc","parameters":{"macroParams":{"minLevel":{"value":"1"},"maxLevel":{"value":"2"}},"macroMetadata":{"macroId":{"value":"a5e2968003747a832698e875be764e3e"},"schemaVersion":{"value":"1"},"title":"Table of Contents"}},"localId":"8a35793d-a0f5-4a6d-8d2f-b38adc389464"}},{"type":"heading","attrs":{"level":1},"content":[{"text":"Introduction","type":"text"}]},{"type":"paragraph","content":[{"text":"To enhance the capabilities of Dataflow it is possible to develop custom stages that can be used in the pipelines you create with Dataflow itself in the same way as any of Dataflow's own components.","type":"text"}]},{"type":"paragraph","content":[{"text":"This article describes how you can do automatic tests with JUnit to test the custom stages you develop.","type":"text"}]},{"type":"paragraph","content":[{"text":"To show examples, we will use a project whose full code can be found on github. This project includes test examples for the JDBC components of Dataflow itself. In this way it shows examples for both source type, target type and processor type stages.","type":"text"}]},{"type":"heading","attrs":{"level":1},"content":[{"text":"Project Configuration","type":"text"}]},{"type":"paragraph","content":[{"text":"The example project on ","type":"text"},{"text":"github","type":"text","marks":[{"type":"link","attrs":{"href":"https://github.com/onesaitplatform/onesaitplatform-dataflow-toolbox/tree/main/test-examples/stages"}}]},{"text":", uses Maven for dependency management. The dependencies used are described below.","type":"text"}]},{"type":"paragraph","content":[{"text":"Versions used:","type":"text"}]},{"type":"codeBlock","content":[{"text":"\n\t1.8\n\t1.8\n\t3.23.0\n\t5.9.3\n\t2.2.220\n","type":"text"}]},{"type":"paragraph","content":[{"text":"JUnit library, in this case, using JUnit 5.","type":"text"}]},{"type":"codeBlock","content":[{"text":"\n org.junit.jupiter\n\tjunit-jupiter-api\n\t${junit.version}\n\ttest\n","type":"text"}]},{"type":"paragraph","content":[{"text":"In this case, since we are testing the Dataflow JDBC library, we need to declare the library that contains the components we are going to test. In a component development project, this declaration would not be necessary as the tests would be part of the project itself.","type":"text"}]},{"type":"codeBlock","content":[{"text":"\n com.streamsets\n\tstreamsets-datacollector-stagesupport\n\t${sdc.version}\n\ttest\n","type":"text"}]},{"type":"paragraph","content":[{"text":"In many cases, we will need additional libraries to test external connections. The type of test—whether unit tests with mocks or integration tests with external systems—will determine this need. In this example, we will use a database to test the connections directly. For this purpose, we use the H2 in-memory database.","type":"text"}]},{"type":"codeBlock","content":[{"text":"\n com.h2database\n\th2\n\t${h2.version}\n\ttest\n","type":"text"}]},{"type":"paragraph","content":[{"text":"Finally, it is necessary to declare the specific Dataflow dependencies that will be used to execute the stages under test and to verify the results.","type":"text"}]},{"type":"codeBlock","attrs":{"language":"xml"},"content":[{"text":"\n\tcom.streamsets\n\tstreamsets-datacollector-api\n\t${sdc.version}\n\ttest\n\n\n\tcom.streamsets\n\tstreamsets-datacollector-sdk\n\t${sdc.version}\n\ttest\n","type":"text"}]},{"type":"heading","attrs":{"level":1},"content":[{"text":"Source Test","type":"text"}]},{"type":"paragraph","content":[{"text":"In this example, to test the JDBC source, we start an H2 database and populate it with some test tables.","type":"text"}]},{"type":"codeBlock","attrs":{"language":"java"},"content":[{"text":"@BeforeAll\npublic static void setUp() throws SQLException {\n\tfinal String sdcId = \"testingID\";\n\tUtils.setSdcIdCallable(new Callable() {\n\t\t@Override\n\t\tpublic String call() {\n\t\t\treturn sdcId;\n\t\t}\n\t});\n\n\t// Create a table in H2 and put some data in it for querying.\n\tconnection = DriverManager.getConnection(h2ConnectionString, username, password);\n\ttry (Statement statement = connection.createStatement()) {\n\t\t// Setup table\n\t\tstatement.addBatch(\"CREATE SCHEMA IF NOT EXISTS TEST;\");\n\t\tstatement.addBatch(\n\t\t\t\t\"CREATE TABLE IF NOT EXISTS \"+fullTableName+\" \" +\n\t\t\t\t\t\t\"(p_id INT NOT NULL, first_name VARCHAR(255), last_name VARCHAR(255));\"\n\t\t\t\t);\n\n\t\t// Add some data\n\t\tstatement.addBatch(\"INSERT INTO \"+fullTableName+\" VALUES (1, 'Adam', 'Kunicki')\");\n\t\tstatement.addBatch(\"INSERT INTO \"+fullTableName+\" VALUES (2, 'Jon', 'Natkins')\");\n\t\tstatement.addBatch(\"INSERT INTO \"+fullTableName+\" VALUES (3, 'Jon', 'Daulton')\");\n\t\tstatement.addBatch(\"INSERT INTO \"+fullTableName+\" VALUES (4, 'Girish', 'Pancha')\");\n\n\t\tstatement.executeBatch();\n\t}\n}","type":"text"}]},{"type":"paragraph","content":[{"text":"Similarly, at the end of the test, we clean up the database.","type":"text"}]},{"type":"codeBlock","attrs":{"language":"java"},"content":[{"text":"@AfterAll\npublic static void tearDown() throws SQLException {\n\ttry (Statement statement = connection.createStatement()) {\n\t\t// Setup table\n\t\tstatement.execute(\"DROP TABLE IF EXISTS \"+fullTableName+\";\");\n\t\tstatement.execute(\"DROP SCHEMA IF EXISTS \"+schema+\";\");\n\t}\n\n\t// Last open connection terminates H2\n\tconnection.close();\n}","type":"text"}]},{"type":"paragraph","content":[{"text":"To conduct the test, the first thing we do is create a JDBC source component.","type":"text"}]},{"type":"codeBlock","attrs":{"language":"java"},"content":[{"text":"JdbcSource origin = new JdbcSource(\n\t\t\t\ttrue,\n\t\t\t\tquery,\n\t\t\t\tinitialOffset,\n\t\t\t\t\"P_ID\",\n\t\t\t\tfalse,\n\t\t\t\t\"\",\n\t\t\t\t1000,\n\t\t\t\tJdbcRecordType.LIST_MAP,\n\t\t\t\tnew CommonSourceConfigBean(queriesPerSecond, BATCH_SIZE, CLOB_SIZE, CLOB_SIZE),\n\t\t\t\tfalse,\n\t\t\t\t\"\",\n\t\t\t\tcreateConfigBean(h2ConnectionString, username, password),\n\t\t\t\tUnknownTypeAction.STOP_PIPELINE,\n\t\t\t\tqueryInterval\n\t\t\t\t);","type":"text"}]},{"type":"paragraph","content":[{"text":"The initialization values depend on the type of component, so those used in this example are specific to the JDBC source.","type":"text"}]},{"type":"paragraph","content":[{"text":"Next, we create a ","type":"text"},{"text":"SourceRunner","type":"text","marks":[{"type":"code"}]},{"text":" object, which will allow us to execute the \"reads\" from the source.","type":"text"}]},{"type":"codeBlock","content":[{"text":"SourceRunner runner = new SourceRunner.Builder(JdbcDSource.class, origin)\n\t\t.addOutputLane(\"lane\")\n\t\t.build();\n\nrunner.runInit();","type":"text"}]},{"type":"paragraph","content":[{"text":"After this, we simulate a sequence of reads (and writes, if needed) from the database using the source. Additionally, we ensure to release the runner's resources once the test execution is complete.","type":"text"}]},{"type":"codeBlock","content":[{"text":"try {\n\t// Check that existing rows are loaded.\n\tStageRunner.Output output = runner.runProduce(null, 2);\n\tMap> recordMap = output.getRecords();\n\tList parsedRecords = recordMap.get(\"lane\");\n\n\tassertEquals(2, parsedRecords.size());\n\tassertEquals(\"2\", output.getNewOffset());\n\n\t// Check that the remaining rows in the initial cursor are read.\n\toutput = runner.runProduce(output.getNewOffset(), 100);\n\tparsedRecords = output.getRecords().get(\"lane\");\n\tassertEquals(2, parsedRecords.size());\n\n\t// Check that new rows are loaded.\n\trunInsertNewRows();\n\toutput = runner.runProduce(output.getNewOffset(), 100);\n\tparsedRecords = output.getRecords().get(\"lane\");\n\n\tassertEquals(2, parsedRecords.size());\n\tassertEquals(\"10\", output.getNewOffset());\n\n\t// Check that older rows are not loaded.\n\trunInsertOldRows();\n\toutput = runner.runProduce(output.getNewOffset(), 100);\n\tparsedRecords = output.getRecords().get(\"lane\");\n\tassertEquals(0, parsedRecords.size());\n} finally {\n\trunner.runDestroy();\n}","type":"text"}]},{"type":"paragraph","content":[{"text":"There's no need to detail the test as it is specific to the component we are testing in the example. The important thing is to see how the ","type":"text"},{"text":"SourceRunner","type":"text","marks":[{"type":"code"}]},{"text":" object (runner in the example) is used to produce records. It is also useful to note how the values of the obtained data can be checked to ensure they are as expected.","type":"text"}]},{"type":"paragraph","content":[{"text":"Complete example on ","type":"text"},{"text":"github","type":"text","marks":[{"type":"link","attrs":{"href":"https://github.com/onesaitplatform/onesaitplatform-dataflow-toolbox/tree/main/test-examples/stages"}}]},{"text":".","type":"text"}]},{"type":"heading","attrs":{"level":1},"content":[{"text":"Destination Test","type":"text"}]},{"type":"paragraph","content":[{"text":"To test writing to the destination, a strategy similar to the one used for testing reading was employed. Therefore, we won't repeat the similar code for setting up the database. For the complete example, please refer to ","type":"text"},{"text":"github","type":"text","marks":[{"type":"link","attrs":{"href":"https://github.com/onesaitplatform/onesaitplatform-dataflow-toolbox/tree/main/test-examples/stages"}}]},{"text":". The main difference is that a destination object, JdbcTarget, will be created in our example:","type":"text"}]},{"type":"codeBlock","content":[{"text":"Target target = new JdbcTarget(\n\t\tschema, \n\t\ttableName, \n\t\tfieldMappings, \n\t\tfalse, \n\t\tfalse, \n\t\tfalse,\n\t\tJdbcMultiRowRecordWriter.UNLIMITED_PARAMETERS, \n\t\tChangeLogFormat.NONE, \n\t\tJDBCOperationType.INSERT,\n\t\tUnsupportedOperationAction.DISCARD, \n\t\tcreateConfigBean(h2ConnectionString, username, password),\n\t\tCollections.emptyList());","type":"text"}]},{"type":"paragraph","content":[{"text":"A ","type":"text"},{"text":"TargetRunner","type":"text","marks":[{"type":"code"}]},{"text":" is created to execute the write operation. To write data, you need to create the records that will form the write batch. Finally, the ","type":"text"},{"text":"TargetRunner","type":"text","marks":[{"type":"code"}]},{"text":" is used to write these records to the destination.","type":"text"}]},{"type":"codeBlock","content":[{"text":"TargetRunner targetRunner = new TargetRunner.Builder(JdbcDTarget.class, target).build();\n\nRecord record = RecordCreator.create();\nList fields = new ArrayList<>();\nfields.add(Field.create(1));\nfields.add(Field.create(\"Adam\"));\nfields.add(Field.create(\"Kunicki\"));\nfields.add(Field.createDatetime(new Instant().toDate()));\nrecord.set(Field.create(fields));\n\nList singleRecord = Collections.unmodifiableList(Arrays.asList(record));\ntargetRunner.runInit();\ntargetRunner.runWrite(singleRecord);","type":"text"}]},{"type":"paragraph","content":[{"text":"Finally, you need to verify that the data has been written to the destination. In this example, we check if the data has been written to the database.","type":"text"}]},{"type":"codeBlock","content":[{"text":"connection = DriverManager.getConnection(h2ConnectionString, username, password);\ntry (Statement statement = connection.createStatement()) {\n\tResultSet rs = statement.executeQuery(\"SELECT COUNT(*) FROM \" + fullTableName);\n\trs.next();\n\tassertEquals(1, rs.getInt(1));\n}","type":"text"}]},{"type":"heading","attrs":{"level":1},"content":[{"text":"Processor Test","type":"text"}]},{"type":"paragraph","content":[{"text":"This case is very similar to the Sources and Destinations tests. The main difference is that a processor component and a ","type":"text"},{"text":"ProcessorRunner","type":"text","marks":[{"type":"code"}]},{"text":" are used.","type":"text"}]},{"type":"codeBlock","content":[{"text":"JdbcLookupDProcessor processor = createProcessor();\n\nProcessorRunner processorRunner = new ProcessorRunner.Builder(JdbcLookupDProcessor.class, processor)\n\t\t.addConfiguration(\"query\", listQuery).addConfiguration(\"columnMappings\", columnMappings)\n\t\t.addConfiguration(\"multipleValuesBehavior\", MultipleValuesBehavior.FIRST_ONLY)\n\t\t.addConfiguration(\"missingValuesBehavior\", MissingValuesBehavior.SEND_TO_ERROR)\n\t\t.addConfiguration(\"maxClobSize\", 1000).addConfiguration(\"maxBlobSize\", 1000)\n\t\t.addConfiguration(\"validateColumnMappings\", true).addOutputLane(\"lane\").build();","type":"text"}]},{"type":"paragraph","content":[{"text":"The remaining checks are similar to those for sources and destinations. You can view the complete example on ","type":"text"},{"text":"github","type":"text","marks":[{"type":"link","attrs":{"href":"https://github.com/onesaitplatform/onesaitplatform-dataflow-toolbox/tree/main/test-examples/stages"}}]},{"text":".","type":"text"}]},{"type":"paragraph"}],"version":1}

Browser not supported