WonderBeat

"Flink-RocksDBStateDataTransfer-thread-1" #570511 daemon prio=5 os_prio=0 cpu=40891.60ms elapsed=80173.33s tid=0x00007fb857d21000 nid=0x98839 waiting on condition  [0x00007fb6b56f2000]
   java.lang.Thread.State: TIMED_WAITING (sleeping)
	at java.lang.Thread.sleep(java.base@11.0.18/Native Method)
	at org.apache.hadoop.fs.s3a.Invoker.retryUntranslated(Invoker.java:441)
	at org.apache.hadoop.fs.s3a.Invoker.retry(Invoker.java:318)
	at org.apache.hadoop.fs.s3a.WriteOperationHelper.finalizeMultipartUpload(WriteOperationHelper.java:353)
	at org.apache.hadoop.fs.s3a.WriteOperationHelper.completeMPUwithRetries(WriteOperationHelper.java:396)
	at org.apache.hadoop.fs.s3a.S3ABlockOutputStream$MultiPartUpload.lambda$complete$1(S3ABlockOutputStream.java:886)
	at org.apache.hadoop.fs.s3a.S3ABlockOutputStream$MultiPartUpload$$Lambda$2906/0x000000084099c440.apply(Unknown Source)
	at org.apache.hadoop.fs.statistics.impl.IOStatisticsBinding.trackDurationOfInvocation(IOStatisticsBinding.java:464)

WonderBeat

Job Manager

2022-12-27 13:38:47.155 StaticFileSplitEnumerator  - Assigned split to subtask 1 : FileSourceSplit: s3a://bucket/2000000/part-00001-00cb73ef-346b-4e1e-a86a-007223ddf275-c000.zstd.parquet [0, 97489087)  hosts=[localhost] ID=0000000032 position=null
2022-12-27 13:38:47.156 StaticFileSplitEnumerator  - Assigned split to subtask 9 : FileSourceSplit: s3a://bucket/2000000/part-00002-00cb73ef-346b-4e1e-a86a-007223ddf275-c000.zstd.parquet [0, 97342071)  hosts=[localhost] ID=0000000033 position=null
2022-12-27 13:38:47.156 StaticFileSplitEnumerator  - Assigned split to subtask 6 : FileSourceSplit: s3a://bucket/2000000/part-00000-00cb73ef-346b-4e1e-a86a-007223ddf275-c000.zstd.parquet [0, 97377047)  hosts=[localhost] ID=0000000031 position=null
2022-12-27 13:38:47.157 StaticFileSplitEnumerator  - Assigned split to subtask 5 : FileSourceSplit: s3a://bucket/2000000/part-00003-00cb73ef-346b-4e1e-a86a-007223ddf275-c000.zstd.parquet [0, 97406878)  hosts=[localhost] ID=0000000034 position=null
2022-12-27 1

WonderBeat

TrackingFsDataInputStream batch tracking issue

org.apache.flink.connector.file.src.impl.StreamFormatAdapter.TrackingFsDataInputStream wraps underlying InputStream to count bytes consumed. org.apache.flink.connector.file.src.impl.StreamFormatAdapter.Reader relies on this to create batches of data.

            while (stream.hasRemainingInBatch() && (next = reader.read()) != null) {
                result.add(next);
            }

WonderBeat

Letter	Encryption
a	cd86
b	cea8
c	cfc8
d	c8c7
e	c9fb
f	ca47
g	cb4e
h	c446

WonderBeat

connector.class=com.github.jcustenborder.kafka.connect.spooldir.SpoolDirCsvSourceConnector
csv.first.row.as.header=false
finished.path=/tmp/spooldir_finished
tasks.max=1
halt.on.error=true
schema.generation.enabled=false
value.converter.schema.registry.url=https://ip-10-0-0-24.eu-west-1.compute.internal:8481,https://ip-10-0-0-25.eu-west-1.compute.internal:8481
input.file.pattern=.*.csv
name=direct-mail-csv
topic=csv-test

WonderBeat

public interface TokenManager {
    enum VerificationResult {
        OK, INVALID;
    }

    byte[] generate();

    VerificationResult verify(byte[] token);

}

WonderBeat

import com.google.common.base.Preconditions;

interface TokenGenerator {
    enum VerificationResult { OK, INVALID }

    byte[] generate();
    VerificationResult verify(byte[] token);

}

WonderBeat

val flowEventBus = EventBus.create(new RingBufferDispatcher("flow-bus", 64))
  val inIOEventBus = EventBus.create(new ThreadPoolExecutorDispatcher(15, 32, "io-input-bus"))
  val outIOEventBus = EventBus.create(new ThreadPoolExecutorDispatcher(15, 32, "io-output-bus"))

  inIOEventBus.on(Selectors.`type`(ReadKafka.getClass), new Consumer[Event[Ticket]] {
    override def accept(event: Event[Ticket]): Unit = {
      val ticket = event.getData
      val writeTask = new WriteTask(ticket.reader.read(ticket.offset, ticket.batchSize), ticket.writer)
      flowEventBus.notify(WriteKafkaRequest, Event.wrap(writeTask))
    }

WonderBeat

1234
1111
1337
0000
1984
6969
5555
1212
7777
6666

WonderBeat

public class Main {
  /**
   * 7
   **/
    public static int method_got7(int a) {
        switch(a) {
            case 1: return 2;
            case 2: return 3;
            case 3: return 4;
            case 4: return 5;