public static void main(String[] args) throws Exception {
    Builder builder = Builder.newBuilder();

    // Requests from the "quiet" bank branch (high throttling).
    Streamlet<WireRequest> quietBranch = builder.newSource(() -> new WireRequest(20))
        .setNumPartitions(1)
        .setName("quiet-branch-requests")
        .filter(WireRequestsTopology::checkRequestAmount)
        .setName("quiet-branch-check-balance");

    // Requests from the "medium" bank branch (medium throttling).
    Streamlet<WireRequest> mediumBranch = builder.newSource(() -> new WireRequest(10))
        .setNumPartitions(2)
        .setName("medium-branch-requests")
        .filter(WireRequestsTopology::checkRequestAmount)
        .setName("medium-branch-check-balance");

    // Requests from the "busy" bank branch (low throttling).
    Streamlet<WireRequest> busyBranch = builder.newSource(() -> new WireRequest(5))
        .setNumPartitions(4)
        .setName("busy-branch-requests")
        .filter(WireRequestsTopology::checkRequestAmount)
        .setName("busy-branch-check-balance");

    // Here, the streamlets for the three bank branches are united into one. The fraud
    // detection filter then operates on that unified streamlet.
    quietBranch
        .union(mediumBranch)
        .setNumPartitions(2)
        .setName("union-1")
        .union(busyBranch)
        .setName("union-2")
        .setNumPartitions(4)
        .filter(WireRequestsTopology::fraudDetect)
        .setName("all-branches-fraud-detect")
        .log();

    Config config = Config.newBuilder()
        .setDeliverySemantics(Config.DeliverySemantics.EFFECTIVELY_ONCE)
        .setNumContainers(2)
        .build();

    // Fetches the topology name from the first command-line argument
    String topologyName = StreamletUtils.getTopologyName(args);

    // Finally, the processing graph and configuration are passed to the Runner, which converts
    // the graph into a Heron topology that can be run in a Heron cluster.
    new Runner().run(topologyName, config, builder);
  }
public static void main ( String [ ] args ) throws Exception { Builder builder = Builder . newBuilder ( ) ; Streamlet < WireRequest > quietBranch = builder . newSource ( ( ) -> new WireRequest ( 20 ) ) . setNumPartitions ( 1 ) . setName ( "quiet-branch-requests" ) . filter ( WireRequestsTopology :: checkRequestAmount ) . setName ( "quiet-branch-check-balance" ) ; Streamlet < WireRequest > mediumBranch = builder . newSource ( ( ) -> new WireRequest ( 10 ) ) . setNumPartitions ( 2 ) . setName ( "medium-branch-requests" ) . filter ( WireRequestsTopology :: checkRequestAmount ) . setName ( "medium-branch-check-balance" ) ; Streamlet < WireRequest > busyBranch = builder . newSource ( ( ) -> new WireRequest ( 5 ) ) . setNumPartitions ( 4 ) . setName ( "busy-branch-requests" ) . filter ( WireRequestsTopology :: checkRequestAmount ) . setName ( "busy-branch-check-balance" ) ; quietBranch . union ( mediumBranch ) . setNumPartitions ( 2 ) . setName ( "union-1" ) . union ( busyBranch ) . setName ( "union-2" ) . setNumPartitions ( 4 ) . filter ( WireRequestsTopology :: fraudDetect ) . setName ( "all-branches-fraud-detect" ) . log ( ) ; Config config = Config . newBuilder ( ) . setDeliverySemantics ( Config . DeliverySemantics . EFFECTIVELY_ONCE ) . setNumContainers ( 2 ) . build ( ) ; String topologyName = StreamletUtils . getTopologyName ( args ) ; new Runner ( ) . run ( topologyName , config , builder ) ; }
All Heron topologies require a main function that defines the topology's behavior
at runtime