NajeebArif

//...

@Override
@SneakyThrows
public CreatedOrder createOrder(Double totalAmount, URI returnUrl) {
    final OrderRequest orderRequest = createOrderRequest(totalAmount, returnUrl);
    final OrdersCreateRequest ordersCreateRequest = new OrdersCreateRequest().requestBody(orderRequest);
    final HttpResponse<Order> orderHttpResponse = payPalHttpClient.execute(ordersCreateRequest);
    final Order order = orderHttpResponse.result();
    LinkDescription approveUri = extractApprovalLink(order);

NajeebArif

//...

@PostMapping
public String placeOrder(@RequestParam Double totalAmount, HttpServletRequest request){
    final URI returnUrl = buildReturnUrl(request);
    CreatedOrder createdOrder = paymentService.createOrder(totalAmount, returnUrl);
    return "redirect:"+createdOrder.getApprovalLink();
}

//...

NajeebArif

//...

private String orderId = "";

@GetMapping
public String orderPage(Model model){
    model.addAttribute("orderId",orderId);
    return "order";
}

NajeebArif

//...

@Override
@SneakyThrows
public void captureOrder(String orderId) {
    final OrdersCaptureRequest ordersCaptureRequest = new OrdersCaptureRequest(orderId);
    final HttpResponse<Order> httpResponse = payPalHttpClient.execute(ordersCaptureRequest);
    log.info("Order Capture Status: {}",httpResponse.result().status());
}

NajeebArif

package narif.poc.springkstreampoc;

import lombok.extern.slf4j.Slf4j;
import narif.poc.springkstreampoc.exceptions.InvalidCreditCardException;
import org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.StreamsBuilderFactoryBean;
import org.springframework.kafka.config.StreamsBuilderFactoryBeanConfigurer;

NajeebArif

package narif.poc.springkstreampoc;

import narif.poc.springkstreampoc.model.OrderInputMsg;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.junit.jupiter.api.Test;

import java.util.Properties;
import java.util.UUID;

NajeebArif

package narif.poc.springkstreampoc;

import lombok.extern.slf4j.Slf4j;
import narif.poc.springkstreampoc.model.OrderInputMsg;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.ValueMapper;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;

NajeebArif

spring:
  cloud:
    stream:
      bindings:
        orderProcessor-in-0:
          destination: raw-order-topic
        orderProcessor-out-0:
          destination: sane-order-topic
        upperCaseProcessor-in-0:

NajeebArif

package narif.poc.springkstreampoc;

import lombok.extern.slf4j.Slf4j;
import narif.poc.springkstreampoc.model.OrderInputMsg;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Transformer;
import org.apache.kafka.streams.kstream.ValueMapper;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.springframework.boot.SpringApplication;

NajeebArif

2022-05-17 02:43:20.772  INFO 7705 --- [-StreamThread-1] n.p.s.SpringKStreamPocApplication        : Order input msg received with key: 157ff176-88c7-4dde-8d05-d80b4d14ff3c and payload: OrderInputMsg(orderId=6615e92f-fb23-4ebd-8a46-9f89bf8b71cf, itemName=PS5 0, userName=Najeeb, creditCardNumber=1111-2222-3333-5555, orderAmount=1000.0)
2022-05-17 02:43:20.778  INFO 7705 --- [-StreamThread-1] n.p.s.SpringKStreamPocApplication        : ORDER CREDIT CARD INFO MASKED FOR KEY: 157ff176-88c7-4dde-8d05-d80b4d14ff3c, VALUE:OrderInputMsg(orderId=6615e92f-fb23-4ebd-8a46-9f89bf8b71cf, itemName=PS5 0, userName=Najeeb, creditCardNumber=XXXX-XXXX-XXXX-5555, orderAmount=1000.0)
2022-05-17 02:43:20.791  INFO 7705 --- [read-1-producer] org.apache.kafka.clients.Metadata        : [Producer clientId=OrderProcessorStream-8b163842-a326-4f42-acc3-32301b6f65b8-StreamThread-1-producer] Resetting the last seen epoch of partition sane-order-topic-0 to 0 since the associated topicId changed from null to sxmI1cSKSLOz01BUdQ68nw
2022-05-17 0