KafkaToS3WithPaquet

    /* ProduceAndConsumeToS3AreOK 테스트 시나리오

+--------------+        +--------------+       +--------------+      +--------------+     +--------------+
|              |        |              |       |              |      |              |     |              |
|  Kafka       |        |  Kafka       |       |   Flow       |      | S3 Upload    |     | S3 DownLoad  |
|  (Producer)  |------->|  (Consumer)  |------>|   Convert    |----->|              |---->|              |
|              |        |              |       |   Pqrquet    |      |              |     |              |
|              |        |              |       |              |      |              |     |              |
+--------------+        +--------------+       +--------------+      +--------------+     +--------------+

  JSON 이벤트를 N개 발생  ->    수신검증    ->     N개씩 파일묶음처리   ->    N개파일 업로드검증  ->  N개파일 다운로드검증및 데이터 변환검증

     */

Source<String, NotUsed> sourceJson = Source.range(1, testCount)
        .map(number -> {
                // JSON 형태의 다양한 소스
                S3TestJsonModel s3TestJsonModel = new S3TestJsonModel();
                s3TestJsonModel.count = number;
                String jsonOriginData = mapper.writeValueAsString(s3TestJsonModel);

                // 원본 유지를 위한 정의 모델
                S3TestModel s3TestModel = new S3TestModel();
                s3TestModel.jsonValue = jsonOriginData;
                return mapper.writeValueAsString(s3TestModel);
        });

Flow<String, ProducerRecord<String, String>, NotUsed> flowProducer = Flow.of(String.class)
        .map(value -> new ProducerRecord<>(testTopicName, testKey, value));

Sink<ProducerRecord<String, String>, CompletionStage<Done>> sinkProducer = Producer.plainSink(producerSettings);

sourceJson.via(flowProducer).toMat(sinkProducer, Keep.right()).run(materializer);

public class S3TestModel {

    public String name = "S3TestModel";

    public String version = "0.0.1";

    public String jsonValue;
}

public class S3TestJsonChildModel {

    public String name = "S3TestJsonChildModel";

    public int count = 2;

    public String subData = "SomeData";
}

Consumer
        .plainSource(
                consumerSettings,
                Subscriptions.topics(testTopicName))
        .grouped(maxEntityPerFile)
        .to(Sink.foreach(group -> {

                S3TestModel model = new S3TestModel();
                model.jsonValue = "example data";

                Schema schema = new Schema.Parser().parse(schemaString);

                // Avro GenericRecord 생성
                GenericRecord record = new GenericData.Record(schema);

                curFileIdx[0]++;

                String dynamicFileKey = fileName + curFileIdx[0];

                group.forEach(msg -> {
                try {
                        // AnyJson을 처리할수 있는~ S3TestModel
                        S3TestModel obj = mapper.readValue(msg.value(), S3TestModel.class);
                        record.put("name", obj.name);
                        record.put("jsonValue", obj.jsonValue);

                } catch (JsonProcessingException e) {
                        throw new RuntimeException(e);
                }
                debugKafkaMsgAndConfirm(msg.key(), msg.value(), confirmActor, testKey, "consumer1");
                });
        .................

Source.single(byteString)
        .runWith(S3.multipartUpload(bucketName, dynamicFileKey)
                .withAttributes(S3Attributes.settings(s3Settings)), materializer)
        .thenAccept(result -> {
                LocalTime now = LocalTime.now();
                DateTimeFormatter formatter = DateTimeFormatter.ofPattern("HH:mm:ss");
                String formatedNow = now.format(formatter);
                System.out.println(formatedNow + " Upload complete: " + result.location());
                confirmActor.tell("s3UploadOK", null);
        })
        .exceptionally(throwable -> {
                System.err.println("Upload failed: " + throwable.getMessage());
                return null;
        });

downloadSource
        .runForeach(index -> {
        String dynamicFileKey = fileName + (index + 1);

        // S3에서 파일 다운로드
        CompletionStage<Optional<Pair<Source<ByteString, NotUsed>, ObjectMetadata>>> download =
                S3.download(bucketName, dynamicFileKey)
                        .withAttributes(S3Attributes.settings(s3Settings))
                        .runWith(Sink.head(), materializer);

        download.thenAccept(optionalSourcePair -> {
                        optionalSourcePair.ifPresent(sourcePair -> {
                        LocalTime now = LocalTime.now();
                        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("HH:mm:ss");
                        String formatedNow = now.format(formatter);
                        Source<ByteString, ?> downloadbyteSource = sourcePair.first();
                        downloadbyteSource.runWith(Sink.foreach(byteString -> System.out.println(formatedNow + " Downloaded: " + dynamicFileKey)), materializer);
                        confirmActor.tell("s3DownloadOK", null);
                        });
                })
                .exceptionally(throwable -> {
                        System.err.println("Download failed for " + dynamicFileKey + ": " + throwable.getMessage());
                        return null;
                });
        }, materializer);                          

// KAFKA 1개의 메시지 수신받았다는 이벤트를 관찰자에게 비동기적으로 알려줌~
if (testKey.equals(key)) confirmActor.tell("kafkaOK", null);

================= 분리되어 작동가능한 Context =================

// Kafka 소비 메시지 전체 개수가 맞는지 관찰자를 통해 수신검증 확인 , 여기서 지정된 시간안에 검증확인이 안되면 실패로 간주~
// 검증확인이 빠르게 되면 빠른 검증코드 자동종료로 진행
for (int i = 0; i < testCount; i++) {
 probe.expectMsg(Duration.ofSeconds(5), "kafkaOK");
}

   retry-settings {
    max-retries = 3
    min-backoff = 200ms
    max-backoff = 10s
    random-factor = 0.0
  } 

# Properties for akka.kafka.ProducerSettings can be
# defined in this section or a configuration section with
# the same layout.
akka.kafka.producer {
  # Config path of Akka Discovery method
  # "akka.discovery" to use the Akka Discovery method configured for the ActorSystem
  discovery-method = akka.discovery

  # Set a service name for use with Akka Discovery
  # https://doc.akka.io/docs/alpakka-kafka/current/discovery.html
  service-name = ""

  # Timeout for getting a reply from the discovery-method lookup
  resolve-timeout = 3 seconds

  # Tuning parameter of how many sends that can run in parallel.
  # In 2.0.0: changed the default from 100 to 10000
  parallelism = 10000

  # Duration to wait for `KafkaProducer.close` to finish.
  close-timeout = 60s

  # Call `KafkaProducer.close` when the stream is shutdown. This is important to override to false
  # when the producer instance is shared across multiple producer stages.
  close-on-producer-stop = true

  # Fully qualified config path which holds the dispatcher configuration
  # to be used by the producer stages. Some blocking may occur.
  # When this value is empty, the dispatcher configured for the stream
  # will be used.
  use-dispatcher = "akka.kafka.default-dispatcher"

  # The time interval to commit a transaction when using the `Transactional.sink` or `Transactional.flow`
  # for exactly-once-semantics processing.
  eos-commit-interval = 100ms

  # Properties defined by org.apache.kafka.clients.producer.ProducerConfig
  # can be defined in this configuration section.
  kafka-clients {
  }
}

akka.kafka.consumer {

  enable.auto.commit = true

  kafka-clients {
    bootstrap.servers = "localhost:9092"
  }
}

akka.kafka.committer {

  # Maximum number of messages in a single commit batch
  max-batch = 1000

  # Maximum interval between commits
  max-interval = 3s

  # Parallelsim for async committing
  parallelism = 100

  # API may change.
  # Delivery of commits to the internal actor
  # WaitForAck: Expect replies for commits, and backpressure the stream if replies do not arrive.
  # SendAndForget: Send off commits to the internal actor without expecting replies (experimental feature since 1.1)
  delivery = WaitForAck

  # API may change.
  # Controls when a `Committable` message is queued to be committed.
  # OffsetFirstObserved: When the offset of a message has been successfully produced.
  # NextOffsetObserved: When the next offset is observed.
  when = OffsetFirstObserved
}

# 이 설정은 유닛테스트를 위한 LocalStack 호환버전이며, AWS-S3이용시 호환은 아래링크 참고 호환을 시킵니다.
# link : https://github.com/akka/alpakka/blob/main/s3/src/main/resources/reference.conf
alpakka.s3 {
  buffer = "memory"
  disk-buffer-path = ""

  # default values for AWS configuration
  aws {
    credentials {
      provider = static
      access-key-id = "test"
      secret-access-key = "test"
    }

    region {
      provider = static
      default-region = "us-east-1"
    }
  }

  path-style-access = true
  access-style = virtual
  endpoint-url = "http://localhost:4567"
  list-bucket-api-version = 2
  validate-object-key = true

  retry-settings {
    max-retries = 3
    min-backoff = 200ms
    max-backoff = 10s
    random-factor = 0.0
  }

  multipart-upload {
    retry-settings = ${alpakka.s3.retry-settings}
  }

  sign-anonymous-requests = true
}

Erik Meijer의 한마디~

"이보게, 브라이언 괴츠, C#,파이선,자바스크립트는 물론 심지어 PHP도 async, await를 지원하고 있다네. 그런 기능이 없는 언어는 자바일뿐이야.

람다를 이용해서 콜백함수를 사용하면 된다고? 천만에 콜백은 최악이야. 도움이 안된다고. 자바 9 버전에 담으려고 하는 걸 다 내려놓고 지금당장

asymc, await부터 넣으라고. 그래야 모두가 행복해질수 있어"