Start Debezium and Handle Events

OutboxPoller.java

...
        @PostConstruct
        void start() {
            executor.execute(engine);
        }
...

        /*
         * Connect to the PG database and invoke the method `handleEvent()` when an
         * event has been polled.
         */
        this.engine = EmbeddedEngine.create()
                .using(debeziumConfig)
                .notifying(this::handleEvent)
                .build();

...

        private void handleEvent(SourceRecord sourceRecord) {
          Struct sourceRecordValue = (Struct) sourceRecord.value();
          Envelope.Operation operation = Operation.forCode((String) sourceRecordValue.get(OPERATION));
  
          //Get only the Create's
          if (operation == Operation.CREATE) {
              Struct after = (Struct) sourceRecordValue.get(AFTER);
  
              Map<String, Object> polledMessage = after.schema().fields().stream()
                      .map(Field::name)
                      .map(fieldName -> Pair.of(fieldName, after.get(fieldName)))
                      .collect(toMap(Pair::getKey, Pair::getValue));
  
              //Publish Message to Kafka.
              kafkaSink.sink(polledMessage);
          }
        }

...