//-------------- CDAP API changes --------------
public interface ApplicationContext<T extends Config> {
  T getConfig();
}

public interface Application<T extends Config> {
  void configure(ApplicationConfigurer configurer, ApplicationContext<T> context);
}
 
public abstract class AbstractApplication<T> implements Application<T extends Config> {
  ...
  protected final ApplicationContext<T> getContext() { return context; }
}
 
//-------------- Example Application --------------
public class MyApp extends AbstractApplication<MyApp.MyConfig> {
 
  public static class MyConfig extends Config {
    @Nullable
    @Description("The name of the stream to read from. Defaults to 'A'.")
    private String stream;
 
    @Nullable
    @Description("The name of the table to write to. Defaults to 'X'.")
    private String table;
 
    @Name("flow")
    private MyFlowConfig flowConfig;
 
    private MyConfig() {
      this.stream = "A";
      this.table = "X";
    }
  }
 
  public void configure() {
    // ApplicationContext now has a method to get a custom config object whose fields will
    // be injected using the values given in the RESTful API
    MyConfig config = getContext().getConfig();
    addStream(new Stream(config.stream));
    createDataset(config.table, Table.class);
    addFlow(new MyFlow(config.stream, config.table, config.flowConfig));
  }
}
 
public class MyFlow implements Flow {
  @Property
  private String stream;
  @Property
  private String table;
  @Property
  private FlowConfig flowConfig;
 
  public static final FlowConfig extends Config {
    private ReaderConfig reader;
    private WriterConfig writer;
  }
 
  MyFlow(String stream, String table, FlowConfig flowConfig) {
    this.stream = stream;
    this.table = table;
    this.flowConfig = flowConfig;
  }
 
  @Override
  public FlowSpecification configure() {
    return FlowSpecification.Builder.with()
      .setName("MyFlow")
      .setDescription("Reads from a stream and writes to a table")
      .withFlowlets()
        .add("reader", new StreamReader(flowConfig.reader))
        .add("writer", new TableWriter(flowConfig.writer))
      .connect()
        .fromStream(stream).to("reader")
        .from("reader").to("writer")
      .build();
  }
} 
 
public class StreamReader extends AbstractFlowlet {
  private OutputEmitter<Put> emitter;
 
  @ProcessInput
  public void process(StreamEvent event) {
  }
}

POST /namespaces/default/artifacts/myapp --data-binary @myapp-1.0.0.jar

GET /namespaces/default/artifacts/myapp/versions/1.0.0
 
{
  "name": "purchase",
  "version": "3.1.0",
  "meta": {
    "created": "1234567890000",
    ...
  },
  "classes": {
    "apps": [
      {
        "className": "co.cask.cdap.examples.myapp.MyApp",
        "properties": {
          "stream": { 
            "name": "stream", 
            "description": "The name of the stream to read from. Defaults to 'A'.", 
            "type": "string", 
            "required": false 
          },
          "table": {
            "name": "table",
            "description": "The name of the table to write to. Defaults to 'X'.",
            "type": "string",
            "required": false,
          },
          "flowConfig": {
            "name": "flow",
            "description": "",
            "type": "config",
            ""            
          }
            "fields": {
            "id": { "name": "id", "type": "long", "required": true },
            "digits": { "name": "phoneNumber", "type": "string", "required": true }
          }
        }
      }
    ],
    "plugins": [
    ]
  }
}

PUT /namespaces/default/apps/myapp -H "X-Archive-Name: myapp-1.0.0.jar" -H "Content-Type: application/octet-stream" --data-binary @myapp-1.0.0.jar

PUT /namespaces/default/apps/myapp2 -H "Content-Type: application/json" -d '{ "artifact": { "name": "myapp", "version": "1.0.0" }, "config": { "stream": "B", "table": "X" } }'

POST /namespaces/default/artifacts/myapp --data-binary @myapp-1.0.1.jar

PUT /namespaces/default/apps/myapp3 -H "Content-Type: application/json" -d '{ "artifact": { "name": "my-app", "version": "1.0.1" }, "config": {"stream": "C", "table": "X"} }'

PUT /namespaces/default/apps/myapp/properties -H "Content-Type: application/json" -d '{ "artifact": { "name":"myapp", "version":"1.0.1" }, "config": { "stream": "A", "table": "X" } }'

PUT /namespaces/default/apps/somePipeline -H "Content-Type: application/json" -d '{ "artifact": { "namespace": "system", "name":"ETLBatch", "version":"3.1.0" }, "config": { ... } }'