Talend Component Kit Developer Guide

public Record createRecord() {
    return factory.newRecordBuilder()
            .withString("name", "Gary")
            .withDateTime("date", ZonedDateTime.of(LocalDateTime.of(2011, 2, 6, 8, 0), ZoneId.of("UTC")))
            .build();
}

public Record createRecord() {
    return factory.newRecordBuilder(myAlreadyBuiltSchemaWithSchemaBuilder)
            .withString("name", "Gary")
            .withDateTime("date", ZonedDateTime.of(LocalDateTime.of(2011, 2, 6, 8, 0), ZoneId.of("UTC")))
            .build();
}

public void print(final Record record) {
    final Schema schema = record.getSchema();
    // log in the natural type
    schema.getEntries()
          .forEach(entry -> System.out.println(record.get(Object.class, entry.getName())));
    // log only strings
    schema.getEntries().stream()
          .filter(e -> e.getType() == Schema.Type.STRING)
          .forEach(entry -> System.out.println(record.getString(entry.getName())));
}

public Record create() {
       final Record address = factory.newRecordBuilder()
               .withString("street", "Prairie aux Ducs")
               .withString("city", "Nantes")
               .withString("country", "FRANCE")
               .build();
       return factory.newRecordBuilder()
                     .withBoolean("active", true)
                     .withInt("age", 33)
                     .withLong("duration", 123459)
                     .withFloat("tolerance", 1.1f)
                     .withDouble("balance", 12.58)
                     .withString("name", "John Doe")
                     .withDateTime("birth", ZonedDateTime.now())
                     .withRecord(
                             factory.newEntryBuilder()
                               .withName("address")
                               .withType(Schema.Type.RECORD)
                               .withComment("The user address")
                               .withElementSchema(address.getSchema())
                               .build(),
                             address)
                     .withArray(
                             factory.newEntryBuilder()
                                       .withName("permissions")
                                       .withType(Schema.Type.ARRAY)
                                       .withElementSchema(factory.newSchemaBuilder(Schema.Type.STRING).build())
                                       .build(),
                             asList("admin", "dev"))
                     .build();
   }

@DiscoverSchema
getSchema(@Option MyDataset dataset) {
    return factory.newSchemaBuilder(Schema.Type.RECORD)
            .withEntry(factory.newEntryBuilder().withName("id").withType(Schema.Type.LONG).build())
            .withEntry(factory.newEntryBuilder().withName("name").withType(Schema.Type.STRING).build())
            .build();
}

@DiscoverSchema
public Schema guessSchema(@Option MyDataset dataset, final MyDataLoaderService myCustomService) {
   return myCustomService.loadFirstData().getRecord().getSchema();
}

{"Value":[
  {"value":"v1"},
  {"value":[]}
  ]
}

@Icon(value = Icon.IconType.CUSTOM, custom = "IconName")

{
  "tcomp::configurationtype::name":"test",
  "tcomp::configurationtype::type":"dataset"
}

{
  "tcomp::configurationtype::name":"test",
  "tcomp::configurationtype::type":"datastore"
}

{
  "tcomp::configurationtype::name":"test",
  "tcomp::configurationtype::type":"datasetDiscovery"
}

{nodes: {
             "idForDstore": { datastore:"datastore data", edges:[id:"idForDset"] },
             "idForDset":   { dataset:"dataset data" }
    }
}

@ElementListener
public MyNewData map(final MyData data) {
    return ...;
}

@ElementListener
public MyNewData map(final Record incomingData) {
    String name = incomingData.getString("name");
    int name = incomingData.getInt("age");
    return ...;
}

// equivalent to (using POJO subclassing)

public class Person {
    private String age;
    private int age;

    // getters/setters
}

@ElementListener
public MyNewData map(final Person person) {
    String name = person.getName();
    int age = person.getAge();
    return ...;
}

@BeforeGroup
public void initBatch() {
    // ...
}

@AfterGroup
public void endBatch() {
    // ...
}

@ElementListener
public void map(final MyData data, @Output final OutputEmitter<MyNewData> output) {
    output.emit(createNewData(data));
}

@ElementListener
public void map(final MyData data,
                @Output final OutputEmitter<MyNewData> main,
                @Output("REJECT") final OutputEmitter<MyNewDataWithError> rejected) {
    if (isRejected(data)) {
        rejected.emit(createNewData(data));
    } else {
        main.emit(createNewData(data));
    }
}

// or

@ElementListener
public MyNewData map(final MyData data,
                    @Output("REJECT") final OutputEmitter<MyNewDataWithError> rejected) {
    if (isSuspicious(data)) {
        rejected.emit(createNewData(data));
        return createNewData(data); // in this case the processing continues but notifies another channel
    }
    return createNewData(data);
}

@ElementListener
public MyNewData map(@Input final MyData data, @Input("input2") final MyData2 data2) {
    return createNewData(data1, data2);
}

@Processor(name = "DocOutput")
public class DocOutput implements Serializable {

    @AfterGroup
    public void onCommit(final Collection<Record> records) {
        // save records
    }
}

@ElementListener
public void store(final MyData data) {
    // ...
}

@RunAtDriver
public void run() {
    ...;
}

files[0].path = /home/user/input1.csv
files[0].format = CSV
files[1].path = /home/user/input2.xml
files[1].format = EXCEL

files[length] = 1

// Map<String, FileOptions>
files.key[0] = first-file
files.value[0].path = /home/user/input1.csv
files.value[0].type = CSV
files.key[1] = second-file
files.value[1].path = /home/user/input2.xml
files.value[1].type = EXCEL

// Map<FileOptions, String>
files.key[0].path = /home/user/input1.csv
files.key[0].type = CSV
files.value[0] = first-file
files.key[1].path = /home/user/input2.xml
files.key[1].type = EXCEL
files.value[1] = second-file

{
  "validation::maxLength":"12.34"
}

{
  "validation::minLength":"12.34"
}

{
  "validation::pattern":"test"
}

{
  "validation::max":"12.34"
}

{
  "validation::min":"12.34"
}

{
  "validation::required":"true"
}

{
  "validation::maxItems":"12.34"
}

{
  "validation::minItems":"12.34"
}

{
  "validation::uniqueItems":"true"
}

package com.mycomponent.components.datastore;

@DataStore("DatastoreA") (1)
@GridLayout({ (2)
    // The generated component layout will display one configuration entry per line.
    // Customize it as much as needed.
    @GridLayout.Row({ "apiurl" }),
    @GridLayout.Row({ "username" }),
    @GridLayout.Row({ "password" })
})
@Documentation("A Datastore made of an API URL, a username, and a password. The password is marked as Credential.") (3)
public class DatastoreA implements Serializable {
    @Option
    @Documentation("")
    private String apiurl;

    @Option
    @Documentation("")
    private String username;

    @Option
    @Credential
    @Documentation("")
    private String password;

    public String getApiurl() {
        return apiurl;
    }

    public DatastoreA setApiurl(String apiurl) {
        this.apiurl = apiurl;
        return this;
    }

    public String getUsername() {
        return Username;
    }

    public DatastoreA setuUsername(String username) {
        this.username = username;
        return this;
    }

    public String getPassword() {
        return password;
    }

    public DatastoreA setPassword(String password) {
        this.password = password;
        return this;
    }
}

package com.datastorevalidation.components.dataset;

@DataSet("DatasetA") (1)
@GridLayout({
    // The generated component layout will display one configuration entry per line.
    // Customize it as much as needed.
    @GridLayout.Row({ "datastore" })
})
@Documentation("A Dataset configuration containing a simple datastore") (2)
public class DatasetA implements Serializable {
    @Option
    @Documentation("Datastore")
    private DatastoreA datastore;

    public DatastoreA getDatastore() {
        return datastore;
    }

    public DatasetA setDatastore(DatastoreA datastore) {
        this.datastore = datastore;
        return this;
    }
}

ComponentFamily.dataset.DatasetA._displayName=Dataset A
ComponentFamily.datastore.DatastoreA._displayName=Datastore A

@Service
public class SomeService {

    @CreateConnection
    public Object createConn(@Option("configuration") SomeDataStore dataStore) throws ComponentException {
        Object connection = null;
        //get conn object by dataStore
        return conn;
    }

    @CloseConnection
    public CloseConnectionObject closeConn() {
        return new CloseConnectionObject() {

            public boolean close() throws ComponentException {
                Object connection = this.getConnection();
                //do close action
                return true;
            }

        };
    }
}

@Version(1)
@Icon(value = Icon.IconType.CUSTOM, custom = "SomeInput")
@PartitionMapper(name = "SomeInput")
@Documentation("the doc")
public class SomeInputMapper implements Serializable {

    @Connection
    SomeConnection conn;

}

{
  "condition::if::evaluationStrategy":"DEFAULT",
  "condition::if::negate":"false",
  "condition::if::target":"test",
  "condition::if::value":"value1,value2"
}

{
  "condition::if::evaluationStrategy::0":"DEFAULT",
  "condition::if::evaluationStrategy::1":"LENGTH",
  "condition::if::negate::0":"false",
  "condition::if::negate::1":"true",
  "condition::if::target::0":"sibling1",
  "condition::if::target::1":"../../other",
  "condition::if::value::0":"value1,value2",
  "condition::if::value::1":"SELECTED",
  "condition::ifs::operator":"AND"
}

<plugin>
  <groupId>org.talend.sdk.component</groupId>
  <artifactId>talend-component-maven-plugin</artifactId>
  <version>${component.version}</version>
  <executions>
    <execution>
      <id>talend-dependencies</id>
      <goals>
        <goal>dependencies</goal>
      </goals>
    </execution>
  </executions>
</plugin>

<plugin>
  <groupId>org.talend.sdk.component</groupId>
  <artifactId>talend-component-maven-plugin</artifactId>
  <version>${component.version}</version>
  <executions>
    <execution>
      <id>talend-scan-descriptor</id>
      <goals>
        <goal>scan-descriptor</goal>
      </goals>
    </execution>
  </executions>
</plugin>

<plugin>
  <groupId>org.talend.sdk.component</groupId>
  <artifactId>talend-component-maven-plugin</artifactId>
  <version>${component.version}</version>
  <executions>
    <execution>
      <id>talend-svg2png</id>
      <goals>
        <goal>svg2png</goal>
      </goals>
    </execution>
  </executions>
</plugin>

<plugin>
  <groupId>org.talend.sdk.component</groupId>
  <artifactId>talend-component-maven-plugin</artifactId>
  <version>${component.version}</version>
  <executions>
    <execution>
      <id>talend-component-validate</id>
      <goals>
        <goal>validate</goal>
      </goals>
    </execution>
  </executions>
</plugin>

<plugin>
  <groupId>org.talend.sdk.component</groupId>
  <artifactId>talend-component-maven-plugin</artifactId>
  <version>${component.version}</version>
  <executions>
    <execution>
      <id>talend-component-documentation</id>
      <goals>
        <goal>asciidoc</goal>
      </goals>
    </execution>
  </executions>
</plugin>

mvn talend-component:car

# for a studio
java -jar mycomponent.car studio-deploy /path/to/my/studio
or
java -jar mycomponent.car studio-deploy --location /path/to/my/studio

# for a m2 provisioning
java -jar mycomponent.car maven-deploy /path/to/.m2/repository
or
java -jar mycomponent.car maven-deploy --location /path/to/.m2/repository

java -jar mycomponent.car deploy-to-nexus --url <nexus url> --repo <repository name> --user <username> --pass <password> --threads <parallel threads number> --dir <temp directory>

--pass "Y0u will \ not G4iess i' ^"

$ mvn talend-component:deploy-in-studio -Dtalend.component.studioHome="<studio_path>"

@Version
@Icon(Icon.IconType.KAFKA)
@Emitter(name = "Input")
@AllArgsConstructor
@Documentation("Kafka Input")
public class KafkaInput extends PTransform<PBegin, PCollection<Record>> { (1)

    private final InputConfiguration configuration;

    private final RecordBuilderFactory builder;

    private final PluginCoderFactory coderFactory;

    private KafkaIO.Read<byte[], byte[]> delegate() {
        final KafkaIO.Read<byte[], byte[]> read = KafkaIO.<byte[], byte[]> read()
                .withBootstrapServers(configuration.getBootstrapServers())
                .withTopics(configuration.getTopics().stream().map(InputConfiguration.Topic::getName).collect(toList()))
                .withKeyDeserializer(ByteArrayDeserializer.class).withValueDeserializer(ByteArrayDeserializer.class);
        if (configuration.getMaxResults() > 0) {
            return read.withMaxNumRecords(configuration.getMaxResults());
        }
        return read;
    }

    @Override (2)
    public PCollection<Record> expand(final PBegin pBegin) {
        final PCollection<KafkaRecord<byte[], byte[]>> kafkaEntries = pBegin.getPipeline().apply(delegate());
        return kafkaEntries.apply(ParDo.of(new BytesToRecord(builder))).setCoder(SchemaRegistryCoder.of()); (3)
    }

    @AllArgsConstructor
    private static class BytesToRecord extends DoFn<KafkaRecord<byte[], byte[]>, Record> {

        private final RecordBuilderFactory builder;

        @ProcessElement
        public void onElement(final ProcessContext context) {
            context.output(toRecord(context.element()));
        }

        private Record toRecord(final KafkaRecord<byte[], byte[]> element) {
            return builder.newRecordBuilder().add("key", element.getKV().getKey())
                    .add("value", element.getKV().getValue()).build();
        }
    }
}

@DataSet
public class MyDataSet {
    // ...
}

@DataStore
public class MyDataStore {
    // ...
}


public class MyComponentConfiguration {
    @Option
    private MyDataSet dataset;

    @Option
    private MyDataStore datastore;
}

public class MappingConfiguration {
    @Option
    private Map<String, String> fieldsMapping;

    @Option
    private boolean ignoreCase;

    //...
}

// ...
public class JdbcDataStore
  implements Serializable {

  @Option
  @Completable("driver")
  private String driver;

  // ...
}

@Completion("driver")
public CompletionList findDrivers() {
    return new CompletionList(findDriverList());
}

@Icon(Icon.IconType.DB_INPUT)
@PartitionMapper(family = "jdbc", name = "input")
public class JdbcPartitionMapper
    implements Serializable {
}

.
├── groupId1
│   └── artifactId1
│       ├── version1
│       │   └── artifactId1-version1.jar
│       └── version2
│           └── artifactId1-version2.jar
└── groupId2
    └── artifactId2
        └── version1
            └── artifactId2-version1.jar

<plugin>
  <groupId>org.apache.maven.plugins</groupId>
  <artifactId>maven-shade-plugin</artifactId>
  <version>3.2.1</version>
  <executions>
    <execution>
      <phase>package</phase>
      <goals>
        <goal>shade</goal>
      </goals>
      <configuration>
        <transformers>
          <transformer implementation="org.talend.sdk.component.container.maven.shade.PluginTransformer">
            <session>${session}</session>
          </transformer>
        </transformers>
      </configuration>
    </execution>
  </executions>
  <dependencies>
    <dependency>
      <groupId>org.talend.sdk.component</groupId>
      <artifactId>nested-maven-repository</artifactId>
      <version>${the.plugin.version}</version>
    </dependency>
  </dependencies>
</plugin>

<plugin>
  <groupId>org.apache.maven.plugins</groupId>
  <artifactId>maven-shade-plugin</artifactId>
  <version>3.2.1</version>
  <configuration>
    <createDependencyReducedPom>false</createDependencyReducedPom>
    <filters>
      <filter>
        <artifact>*:*</artifact>
        <excludes>
          <exclude>META-INF/.SF</exclude>
          <exclude>META-INF/.DSA</exclude>
          <exclude>META-INF/*.RSA</exclude>
        </excludes>
      </filter>
    </filters>
  </configuration>
  <executions>
    <execution>
      <phase>package</phase>
      <goals>
        <goal>shade</goal>
      </goals>
      <configuration>
        <shadedClassifierName>shaded</shadedClassifierName>
        <transformers>
          <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
          <transformer
              implementation="org.talend.sdk.component.container.maven.shade.ContainerDependenciesTransformer">
            <session>${session}</session>
            <userArtifacts>
              <artifact>
                <groupId>org.talend.sdk.component</groupId>
                <artifactId>sample-component</artifactId>
                <version>1.0</version>
                <type>jar</type>
              </artifact>
            </userArtifacts>
          </transformer>
          <transformer implementation="org.talend.sdk.component.container.maven.shade.PluginTransformer">
            <session>${session}</session>
            <userArtifacts>
              <artifact>
                <groupId>org.talend.sdk.component</groupId>
                <artifactId>sample-component</artifactId>
                <version>1.0</version>
                <type>jar</type>
              </artifact>
            </userArtifacts>
          </transformer>
        </transformers>
      </configuration>
    </execution>
  </executions>
  <dependencies>
    <dependency>
      <groupId>org.talend.sdk.component</groupId>
      <artifactId>nested-maven-repository</artifactId>
      <version>${the.version}</version>
    </dependency>
  </dependencies>
</plugin>

public class MyConnectionURITest {
    @Test
    public void checkMySQL() {
        assertTrue(new ConnectionService().isValid("jdbc:mysql://localhost:3306/mysql"));
    }

    @Test
    public void checkOracle() {
        assertTrue(new ConnectionService().isValid("jdbc:oracle:thin:@//myhost:1521/oracle"));
    }
}

@RunWith(Parameterized.class) (1)
public class MyConnectionURITest {

    @Parameterized.Parameters(name = "{0}") (2)
    public static Iterable<String> uris() { (3)
        return asList(
            "jdbc:mysql://localhost:3306/mysql",
            "jdbc:oracle:thin:@//myhost:1521/oracle");
    }

    @Parameterized.Parameter (4)
    public String uri;

    @Test
    public void isValid() { (5)
        assertNotNull(uri);
    }
}

@ParameterizedTest
@ValueSource(strings = { "racecar", "radar", "able was I ere I saw elba" })
void mytest(String currentValue) {
    // do test
}

@ParameterizedTest
@MethodSource("stringProvider")
void mytest(String currentValue) {
    // do test
}

static Stream<String> stringProvider() {
    return Stream.of("foo", "bar");
}

public class MyComponentTest {

    @Rule (1)
    public final SimpleComponentRule components = new SimpleComponentRule("org.talend.sdk.component.mycomponent");

    @Test
    public void produce() {
        Job.components() (2)
             .component("mycomponent","yourcomponentfamily://yourcomponent?"+createComponentConfig())
             .component("collector", "test://collector")
           .connections()
             .from("mycomponent").to("collector")
           .build()
           .run();

        final List<MyRecord> records = components.getCollectedData(MyRecord.class); (3)
        doAssertRecords(records); // depending your test
    }
}

public class SimpleComponentRuleTest {

    @ClassRule
    public static final SimpleComponentRule COMPONENT_FACTORY = new SimpleComponentRule("...");

    @Rule (1)
    public final ServiceInjectionRule injections = new ServiceInjectionRule(COMPONENT_FACTORY, this); (2)

    @Service (3)
    private LocalConfiguration configuration;

    @Service
    private Jsonb jsonb;

    @Test
    public void test() {
        // ...
    }
}

@WithComponents("org.talend.sdk.component.junit.component") (1)
public class ComponentExtensionTest {
    @Injected (2)
    private ComponentsHandler handler;

    @Test
    public void manualMapper() {
        final Mapper mapper = handler.createMapper(Source.class, new Source.Config() {

            {
                values = asList("a", "b");
            }
        });
        assertFalse(mapper.isStream());
        final Input input = mapper.create();
        assertEquals("a", input.next());
        assertEquals("b", input.next());
        assertNull(input.next());
    }
}

@WithComponents("...")
class ComponentExtensionTest {

    @Service (1)
    private LocalConfiguration configuration;

    @Service
    private Jsonb jsonb;

    @Test
    void test() {
        // ...
    }
}

job.property("streaming.maxRecords", 5);

public class MyComponentTest {

    @Rule
    public final SimpleComponentRule components = new SimpleComponentRule("org.talend.sdk.component.mycomponent");

    @Test
    public void produce() {
        components.setInputData(asList(createData(), createData(), createData())); (1)

        Job.components()
             .component("emitter","test://emitter")
             .component("out", "yourcomponentfamily://myoutput?"+createComponentConfig())
           .connections()
              .from("emitter").to("out")
           .build()
           .run();

        assertMyOutputProcessedTheInputData();
    }
}

final MyComponentConfig componentConfig = new MyComponentConfig();
componentConfig.setUser("....");
// .. other inits

final Map<String, String> configuration = configurationByExample(componentConfig);

final String uri = "family://component?" +
    configurationByExample().forInstance(componentConfig).configured().toQueryString();

public class MapperTest {

    @ClassRule
    public static final SimpleComponentRule COMPONENT_FACTORY = new SimpleComponentRule(
            "org.company.talend.component");

    @Test
    public void mapper() {
        final Mapper mapper = COMPONENT_FACTORY.createMapper(MyMapper.class, new Source.Config() {{
            values = asList("a", "b");
        }});
        assertEquals(asList("a", "b"), COMPONENT_FACTORY.collectAsList(String.class, mapper));
    }
}

public class ProcessorTest {

    @ClassRule
    public static final SimpleComponentRule COMPONENT_FACTORY = new SimpleComponentRule(
            "org.company.talend.component");

    @Test
    public void processor() {
        final Processor processor = COMPONENT_FACTORY.createProcessor(Transform.class, null);
        final SimpleComponentRule.Outputs outputs = COMPONENT_FACTORY.collect(processor,
                        new JoinInputFactory().withInput("__default__", asList(new Transform.Record("a"), new Transform.Record("bb")))
                                              .withInput("second", asList(new Transform.Record("1"), new Transform.Record("2")))
                );
        assertEquals(2, outputs.size());
        assertEquals(asList(2, 3), outputs.get(Integer.class, "size"));
        assertEquals(asList("a1", "bb2"), outputs.get(String.class, "value"));
    }
}

public class SparkClusterRuleTest {

    @ClassRule
    public static final SparkClusterRule SPARK = new SparkClusterRule("2.10", "1.6.3", 1);

    @Test
    public void classpathSubmit() throws IOException {
        SPARK.submitClasspath(SubmittableMain.class, getMainArgs());

        // wait for the test to pass
    }
}

@WithSpark
class SparkExtensionTest {

    @SparkInject
    private BaseSpark<?> spark;

    @Test
    void classpathSubmit() throws IOException {
        final File out = new File(jarLocation(SparkClusterRuleTest.class).getParentFile(), "classpathSubmitJunit5.out");
        if (out.exists()) {
            out.delete();
        }
        spark.submitClasspath(SparkClusterRuleTest.SubmittableMain.class, spark.getSparkMaster(), out.getAbsolutePath());

        await().atMost(5, MINUTES).until(
                () -> out.exists() ? Files.readAllLines(out.toPath()).stream().collect(joining("\n")).trim() : null,
                equalTo("b -> 1\na -> 1"));
    }
}

<dependency>
  <groupId>org.awaitility</groupId>
  <artifactId>awaitility</artifactId>
  <version>3.0.0</version>
  <scope>test</scope>
</dependency>

await()
    .atMost(5, MINUTES)
    .until(
        () -> out.exists() ? Files.readAllLines(out.toPath()).stream().collect(joining("\n")).trim() : null,
        equalTo("the expected content of the file"));

public class MyRESTApiTest {
    @ClassRule
    public static final JUnit4HttpApi API = new JUnit4HttpApi();

    @Rule
    public final JUnit4HttpApiPerMethodConfigurator configurator = new JUnit4HttpApiPerMethodConfigurator(API);

    @Test
    public void direct() throws Exception {
        // ... do your requests
    }
}

@HttpApi
class JUnit5HttpApiTest {
    @HttpApiInject
    private HttpApiHandler<?> handler;

    @Test
    void getProxy() throws Exception {
        // .... do your requests
    }
}

public class GoogleTest {
    @ClassRule
    public static final JUnit4HttpApi API = new JUnit4HttpApi();

    @Rule
    public final JUnit4HttpApiPerMethodConfigurator configurator = new JUnit4HttpApiPerMethodConfigurator(API);

    @Test
    public void google() throws Exception {
        assertEquals(HttpURLConnection.HTTP_OK, get("https://google.fr?q=Talend"));
    }

    private int get(final String uri) throws Exception {
        // do the GET request, skipped for brievity
    }
}

@HttpApiName("${class}_${method}")
@RepeatedTest(5)
void run() throws Exception {
    // ...
}

@HttpApiName("${class}_${method}_${displayName}")
@RepeatedTest(5)
void run() throws Exception {
    // ...
}

<dependencies>
  <dependency>
    <groupId>org.talend.sdk.component</groupId>
    <artifactId>component-runtime-junit</artifactId>
    <scope>test</scope>
  </dependency>
  <dependency>
    <groupId>org.junit.jupiter</groupId>
    <artifactId>junit-jupiter-api</artifactId>
    <scope>test</scope>
  </dependency>
  <dependency>
    <groupId>org.jboss.shrinkwrap.resolver</groupId>
    <artifactId>shrinkwrap-resolver-impl-maven</artifactId>
    <version>3.1.4</version>
    <scope>test</scope>
  </dependency>
  <dependency>
    <groupId>org.talend.sdk.component</groupId>
    <artifactId>component-runtime-beam</artifactId>
    <scope>test</scope>
  </dependency>
</dependencies>

@Environment(ContextualEnvironment.class)
@Environment(DirectRunnerEnvironment.class)
class TheComponentTest {
    @EnvironmentalTest
    void testWithStandaloneAndBeamEnvironments() {
        from("myfamily://in?config=xxxx")
            .to("myfamily://out")
            .create()
            .execute();
        // add asserts on the output if needed
    }
}

Job.components()
        .component("employee","db://input")
            .property(GroupKeyProvider.class.getName(),
                 (GroupKeyProvider) context -> context.getData().getString("id")) (1)
        .component("salary", "db://input")
        .component("concat", "transform://concat?separator=;")
    .connections()
        .from("employee").to("concat", "string1")
        .from("salary").to("concat", "string2")
    .build()
    .property(GroupKeyProvider.class.getName(), (2)
                 (GroupKeyProvider) context -> context.getData().getString("employee_id"))
    .run();

GroupKeyProvider keyProvider = context -> {
    if ("employee".equals(context.getComponentId())) {
        return context.getData().getString("id");
    }
    return context.getData().getString("employee_id");
};

public class Main {
    public static void main(final String[] args) {
        final ComponentManager manager = ComponentManager.instance()
        Pipeline pipeline = Pipeline.create();
        //Create beam input from mapper and apply input to pipeline
        pipeline.apply(TalendIO.read(manager.findMapper(manager.findMapper("sample", "reader", 1, new HashMap<String, String>() {{
                    put("fileprefix", "input");
                }}).get()))
                .apply(new ViewsMappingTransform(emptyMap(), "sample")) // prepare it for the output record format (see next part)
        //Create beam processor from talend processor and apply to pipeline
                .apply(TalendIO.write(manager.findProcessor("test", "writer", 1, new HashMap<String, String>() {{
                    put("fileprefix", "output");
                }}).get(), emptyMap()));

        //... run pipeline
    }
}

public class Main {
    public static void main(final String[] args) {
        //Component manager and pipeline initialization...

        //Create beam PTransform from processor and apply input to pipeline
        pipeline.apply(TalendFn.asFn(manager.findProcessor("sample", "mapper", 1, emptyMap())).get())), emptyMap());

        //... run pipeline
    }
}

@Option
@DefaultValue("local_configuration:myfamily.model.key")
private String value;

@Data // from lombok, optional
public class MyConfig {
  @Option
  private String defaultUrl;
}

@Service
public class ConfiguredService {
  @Configuration("myprefix")
  private Supplier<MyConfig> config;
}

@Service
public class ConfiguredComponent {
  public ConfiguredComponent(@Configuration("myprefix") final MyConfig config) {
    // ...
  }
}