haigopi

@Slf4j
public class FluxTest {

    @Test
    public void test() {

        Flux<String> b = Flux.just("Bob cat", "is in the ", "party").delayElements(Duration.ofSeconds(2));

        Flux.first(b)

haigopi

@Slf4j
public class FluxPublishTest {

    @Test
    public void test() {

        Scheduler scheduler = Schedulers.immediate();

        Flux.range(1, 5)
                .map(n -> Mono.just(n).publishOn(scheduler).map(k -> process(k))

haigopi

@Slf4j
public class FluxPublishTest {

    @Test
    public void test() {

        Scheduler scheduler = Schedulers.newParallel("P");

        Flux.range(1, 6)
                .parallel()

haigopi

@Slf4j
public class FluxPublishTest {

    @Test
    public void test() {

        Scheduler scheduler = Schedulers.single();

        Flux.range(1, 6)
                .parallel()

haigopi

@Slf4j
public class FluxPublishTest {

    @Test
    public void test() {

        Scheduler scheduler = Schedulers.elastic();

        Flux.range(1, 6)
                .parallel()

haigopi

$ ./confluent start

Starting zookeeper
zookeeper is [UP]
Starting kafka
kafka is [UP]
Starting schema-registry
schema-registry is [UP]
Starting kafka-rest
kafka-rest is [UP]

haigopi

#=
ModuleCaller:
- Julia version: 1.1.0
- Author: GOPI K KANCHARLA
- Date: 2019-02-25
=#


module Bar

haigopi

#=
ReadData:
- Julia version: 
- Author: GOPI K KANCHARLA
- Date: 2019-02-23
=#


println("typeof(1): ", typeof(1))
println("Sys.WORD_SIZE: ",Sys.WORD_SIZE)

haigopi

        ReceiverOptions<Integer, String> options = receiverOptions.subscription(Collections.singleton(topic))
                .addAssignListener(partitions -> log.info("Partitions Assigned {}", AOUtilities.printPartitions(partitions)))
                .addRevokeListener(partitions -> log.info("Partitions Revoked {}", partitions))
//                .assignment(Collections.singleton(new TopicPartition(topic, 5)))  // <-- ** EASY TEST ONLY **
                .commitInterval(Duration.ZERO)
                .commitBatchSize(0)// <-- interval and batch size:  can dramatically improve the performance. Make sure to test the retries as a side effect.
                .consumerProperty(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");


        KafkaReceiver<Integer, String> receiver = KafkaReceiver.create(options);

haigopi

errorCodes:

  codeDescription:

    application:

        NULL_EMPTY_CHECK: 101 | Application details cannot be Empty

    "[application.applicationType]":