Outbox Nedir?

Transactional (Atomik işlem, işlem bütünlüğü) Outbox deseni, bir uygulamanın durumundaki değişiklikleri hem veri tabanına kaydetmek hem de başka servislere bildirmek için kullanılan tasarım deseni olarak bilinir. Transactional Outbox deseni, değişiklikleri bildirmek için veri tabanı ve mesaj aracısını aynı işlem içinde kullanır. Bu sayede iki işlemin ya aynı anda gerçekleşmesini ya da iki işlemin de gerçekleşmemesini sağlar. Mesajı atıp veri tabanına durumun kaydedilmemesinin ya da veri tabanına durumun kaydedilip, mesajın atılamamasının önüne geçer.

Outbox Deseni Kullanım Senaryoları

Outbox desenine aşağıdaki örnekteki gibi durumlarda ihtiyaç duyarız;

• Kullanıcı oluşturma işlemi sonucu oluşan kaydı, kullanıcı hesabı veri tabanına yazma.
• Hesabın başarıyla oluşturulduğunu bu işlemden haberdar olmak isteyen servislere bildirme.

Dapr (Distributed Application Runtime) sağlamış olduğu konfigürasyon kolaylıklarının yanında outbox deseni desteğini de sunarak, uygulamada yapılan durum değişikliklerinde bu değişikliği dinleyen servisleri bilgilendirebilir.

Outbox Deseni Diyagramı

Transactional Outbox özelliğinin nasıl çalıştığı Şekil 1’de gösterilmektedir:

1. Servis A, bir atomik işlem (transaction) içerisinde modelin durumunu, veri tabanında günceller/kaydeder.
2. Aynı atomik işlem altında ilgili mesaj, mesaj aracısına gönderilir. Mesaj aracısı (message broker) iletiyi başarıyla teslim ettiğinde, atomik işlem tamamlanır ve hem objenin durumu hem de mesaj aynı işlem içerisinde tutarlı bir şekilde işlenir.
3. Mesaj aracısı, iletiyi dinleyenlere (subscriberslara) teslim eder.

Outbox Kullanım Örneği Adımları

Outbox özelliğini kullanabilmek için Dapr’ın versiyonunun 12+ olması gerekmektedir.

1. Öncelikle adresindeki adımlar izlenerek Dapr ve Redis kurulumları yapılır. İlk aşama olarak Dapr’da veri tabanı anlamına gelen state store’un ayarlanması gerekmektedir. Dapr’da Redis ön kurulumlu olarak gelmektedir, yapacağımız örnek için biz de Redis’i veri tabanı olarak kullanacağız.
2. Kullanılacak repository clone edilir.
3. Proje intellij idea’da açılarak maven yaşam döngüsü (lifecycle) bölümünden maven reload (yeniden yükleme) işlemi gerçekleştirilir. Aynı zamanda payment ve shipment klasörleri için de mvn clean install yapılması gerekir.
4. Shipment servisi isteği karşılayan taraf olduğu için dinleyen (subscriber) yapıdadır. Payment servisi isteği gönderen taraf olduğu için yayan (publisher) yapıdadır. Payment servisi için oluşturulan örnek aşağıdaki gibidir.

Payment Application

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
@SpringBootApplication
public class PaymentApplication {
private static final String DAPR_STATE_STORE = "statestore";

public static void main(String[] args) throws Exception {
   try (DaprClient client = new DaprClientBuilder().build()) {
         for (int i = 1; i <= 16; i++) {
                Payment payment = new Payment();
                payment.setOrderId(i);
                payment.setStatus(i % 4 == 0 ? Status.FAILED : Status.SUCCESS);
                payment.setPaymentId(RandomStringUtils.randomAlphanumeric(10));
                // Save state into the state store
                TransactionalStateOperation<Payment> tx = new TransactionalStateOperation<>(TransactionalStateOperation.OperationType.UPSERT, new State<>(String.valueOf(payment.getOrderId()), payment, ""));
                client.executeStateTransaction(DAPR_STATE_STORE, List.of(tx)).block();
                System.out.println("Saving Payment: " + payment.getOrderId());
               TimeUnit.MILLISECONDS.sleep(1000);
          }

Payment işlemini yapabilmek için sürecin atomik işlem olduğunu belirtmemiz gerekmektedir. Bunu TransactionalStateOperation sınıfını kullanarak sağlayabiliriz. Transactional çerçeveye alındığında, durumu ve mesajı yazacak ya da ikisini de yazamayacak duruma getirilmiş olur. Transactional Outbox deseninin temelinde de bu kullanım yatmaktadır.

1. Payment service’i için kullanılacak kaynakları yönetebilmek için dapr-resource klasörü içerisinde pubsub.yaml statestore.yaml dosyaları oluşturulur. Ayrıca modülün ana dizininde uygulamayı çalıştırılmasını daha kolay hale getiren dapr.yaml dosyası eklenmiştir. dapr.yaml dosyası sayesinde komut satırında uzun uzun yapılacak tanımlamalar yerine sadece dapr run -f . yazılması yeterli olacaktır. İlgili konfigürasyon dosyasında çalıştırmak istenilen uygulamanın ihtiyaç duyduğu kaynakların tanımları (resource) ve uygulamayı ayağa kaldırırken kullanılacak komutlar (command) tanımlanmalıdır.

PubSub.yaml

1
2
3
4
5
6
7
8
9
10
11
12
apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: paymentpubsub
spec:
  type: pubsub.redis
  version: v1
  metadata:
    - name: redisHost
      value: localhost:6379
    - name: redisPassword
      value: ""

PubSub.yaml konfigürasyon dosyası, Dapr bileşeni olan “paymentpubsub” adlı pub/sub tipinde bir bileşeni temsil eder. Konfigürasyon dosyasında dikkat edilmesi gereken önemli durumlar şunlardır:

• Paymentpubsub bileşeni, Redis’i mesajlaşma sistemi olarak kullanmaktadır.
• Metadata’da verilen isim uygulama içerisinde verilen isimle aynı olmalıdır.
• Name alanında verilen isim uygulamanın hangi bileşenin kullanılacağını bilmesini sağlar.
• spec’in type’ı alanında yazılan pubsub.redis, “paymentpubsub” bileşeninin bir pubsub mekanizması olduğunu ve Redis olduğunu tanımlamaktadır. Bu alanların alabileceği değerlere Dapr’ın sayfasından ulaşabilirsiniz.
• spec altında bulunan metadata kısmında Redis bileşeninin özelliklerinden bahsedilmektedir. Kafka 6379 portunda çalıştından redisHost alanının değeri olarak localhost:6379 girilmesi gerekmektedir.

StateStore.yaml

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: statestore
spec:
  type: state.redis
  version: v1
  metadata:
    - name: redisHost
      value: localhost:6379
    - name: redisPassword
      value: ""
    - name: outboxPublishPubsub # Required
      value: "paymentpubsub"
    - name: outboxPublishTopic # Required
      value: "paymentStatus"

StateStore.yaml konfigürasyon dosyası, “statestore” adlı bir Dapr bileşenini temsil eder ve “statestore” bileşeni Redis tipindeki state store’u kullanmaktadır. Bu kapsamda dikkat edilmesi gereken önemli durumlar şunlardır:

• metadata’da verilen isim uygulama içerisinde verilen isimle aynı olmalıdır. Uygulama metadata’da verilen isme göre, “statestore” bileşenini kullanacağını bilmektedir.
• spec’in type’ı alanında yazılan state.redis bu bileşenin bir veri tabanı olduğunu ve Redis tipinde olduğunu tanımlamaktadır. Bu alanların alabileceği değerlere Dapr’ın sayfasından ulaşabilirsiniz.
• spec altında bulunan metadata kısmında ise bu state store bileşeninin özelliklerinden bahsedilmektedir. Bu alanın alabileceği değerlere Dapr’ın ilgili sayfasından ulaşabilirsiniz.

Dapr.yaml

1
2
3
4
5
6
7
8
version: 1
common:
  resourcesPath: ./dapr-resource/
apps:
  - appID: payment
    appPort: 8080
    appDirPath: ./
    command: [ "java", "-jar", "target/payment-0.0.1-SNAPSHOT.jar" ]

Dapr.yaml konfigürasyon dosyası, uygulamaların tanımlandığı ve yönetildiği bir yapıyı ifade eder. Bu kapsamda common bölümünde Dapr’ın kullanacağı kaynakların tanımlarının bulunduğu dosyanın adresi verilir, apps bölümde ise uygulamayı ayağa kaldırırken kullanacağı parametreler tanımlanır. dapr run -f . komutu çalıştırıldığında Dapr oluşturduğumuz dapr.yaml dosyasını işlemektedir.

Shipment tarafında da clean ve install işlemlerinin tamamlanması gerekmektedir. Shipment servisi Payment servisinden gelecek bilgiyi dinleyen servistir. Shipment servisindeki kodu incelersek;

Shipment servisi isteği karşılayan taraf olduğu için bir endpoint ve hangi topic’i dinleyeceğinin bilgisine ihtiyaç duyar. Aşağıda tanımlı kod parçasını inceleyecek olursak, POST isteğini dinleyen bir endpoint’in oluşturulduğunu görürüz. Bu endpoint, “/paymentStatus” URL’sine sahiptir ve MediaType.ALL_VALUE türündeki tüm içerik türlerini kabul eder. Shipment servisi endpoint’e gelen CloudEvent verisini alır, bu veriyi bir Payment sınıfına dönüştürür ve ardından gelen Payment nesnesiyle ilgili işlemleri gerçekleştirir. Dönüş olarak ResponseEntity tipinde bir mono nesnesi döndürülür. Hata durumunda ise bir iç hata mesajıyla birlikte INTERNAL_SERVER_ERROR durum kodunu döndürür.

ShipmentController

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
@RestController
public class ShipmentController {

private static final Logger logger = LoggerFactory.getLogger(ShipmentController.class);

@Topic(name = "paymentStatus", pubsubName = "shipmentpubsub")
@PostMapping(path = "/paymentStatus", consumes = MediaType.ALL_VALUE)
public Mono < ResponseEntity > getPaymentStatus(@RequestBody(required = false) CloudEvent < String > cloudEvent) {
  return Mono.fromSupplier(() -> {
    try {
      ObjectMapper objectMapper = new ObjectMapper();
      Payment payment = objectMapper.readValue(cloudEvent.getData(), Payment.class);
      logger.info("Subscriber received: " + payment.getOrderId());
      return ResponseEntity.ok("SUCCESS");
    } catch (Exception e) {
      throw new RuntimeException(e);
    }
  });
}

Shipment servisinde sadece veri okunup herhangi bir veri tabanı kullanımı gerekmediği için sadece mesajlaşma altyapısını yönetebilmek için dapr-resource dizininde pub-sub.yaml konfigürasyonu oluşturulur.

PubSub.yaml

1
2
3
4
5
6
7
8
9
10
11
12
apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: shipmentpubsub
spec:
  type: pubsub.redis
  version: v1
  metadata:
  - name: redisHost
    value: localhost:6379
  - name: redisPassword
    value: ""

PubSub.yaml konfigürasyon dosyası, Dapr bileşeni olan “shipmentpubsub” adlı pub/sub tipinde bir bileşeni temsil eder. Bu bileşen, Redis mesajlaşma sistemini kullanmaktadır.

• metadata’da verilen isim uygulama içerisinde verilen isimle aynı olmalıdır. Uygulama kullanacağı bileşenleri bu alanda tanımlı olan isimlerle eşleştirerek kullanmaktadır.
• spec’in type’ı alanında yazılan pubsub.redis bu bileşenin bir pubsub mekanizması olduğunu ve Redis tipinde olduğunu tanımlamaktadır. Bu alanların alabileceği değerlere Dapr’ın sayfasından ulaşabilirsiniz.
• spec altında bulunan metadata kısmında ise bu Redis bileşeninin özelliklerinden bahsedilmektedir. Kafka 6379 portunda çalıştığından ötürü burada Redis host adresine localhost:6379 tanımı yapılması gereklidir.

Dapr.yaml

1
2
3
4
5
6
7
8
version: 1
common:
  resourcesPath: ./dapr-resource/
apps:
  - appID: shipment
    appPort: 8082
    appDirPath: ./
    command: [ "java", "-jar", "target/shipment-0.0.1-SNAPSHOT.jar" ]

Dapr.yaml konfigürasyon dosyası, uygulamaların tanımlandığı ve yönetildiği bir yapıyı ifade eder. Bu kapsamda common bölümünde Dapr’ın kullanacağı kaynakların tanımlarının bulunduğu dosyanın adresi verilir, apps bölümde ise uygulamayı ayağa kaldırırken kullanacağı parametreler tanımlanır. dapr run -f . komutu çalıştırıldığında Dapr oluşturduğumuz dapr.yaml dosyasını işlemektedir.

Redis Örneği

• Intellij’e Redis veri tabanı bağlantısı kurulması gerekmektedir. Öncelikle data source üzerinden Redis ile database connection’ı kurulmalıdır. Redis konteynırı oluşturulurken girilen bilgileri buradaki connection bilgileri alanlarına girmeliyiz sonrasında “Test connection passed” mesajıyla bağlantının doğru bir şekilde kurulduğu görülmektedir. Bunun sonucunda bağlantı kurulmuş olur ve sonraki aşamaya geçilebilir. Şekil 2 ve Şekil 3’de bu adımları ve çıktılarını görebiliriz.

• Kodda yapılan güncellemelerin uygulamaya yansıtılabilmesi için maven install sürecinin tetiklenmesi gerekmektedir. Çünkü Dapr kaynak kodu değil paketlenmiş olan kodu kullanmaktadır. Bu paketlenmiş olan kodda maven install sonucunda oluşmaktadır. Oluşan paket target dizinindeki paketadı-snapshot.jar isimli dosyaya karşılık gelmektedir.
• Maven install işlemi sonrasında dapr run -f . komutu hem shipment hem de payment için çalıştırılır.
• İki uygulamada ayakta olduğu süre boyunca Redis’i yönettiğimiz arayüzde yenile işlemini tetikleyerek hem mesajların işlendiğini hem de veri tabanına karşılık gelen kayıtları görüntüleyebiliriz.

• Gerçekleştirdiğimiz örnekte Redis hem veri tabanı hem de pub/sub mekanizması olarak kullanılmaktadır. Redis üzerinde hem mesaj işlemleri yönetilir hem de payment servisinde oluşan kayıtları görüntüleyebiliriz.

• Dapr, Outbox desenini uygulayabilmesi için iki farklı stream oluşturur bu streamlerden birini kendi iç sürecini yönetmek için kullanmaktadır. Diğerini ise mesajı topic’e iletme işlevini gerçekleştirmek için kullanmaktadır.

Outbox mekanizması Kafka ile dönüştürülmek istenirse nasıl bir değişiklik yapmamız gerekir? Bu sorunun cevabını bir sonraki bölümde arayacağız.

KAFKA DÖNÜŞÜM AŞAMALARI

Kafka ile Redis arasında bir dönüşüm yapmak için yukarıda uygulanmış olunan aşamaların kullanılması gerekmektedir.

Hem Kafka dönüşümünde kullanılacak hem de uygulamanın genelinde kullanılacak kaynakları yönetebilmek için resource dizini içerisinde pubsub.yaml, dapr.yaml ve statestore.yaml dosyalarının oluşturulması gerekmektedir. Bu dosyalar temelde Redis ve Kafka’ya ait statestore ve pubsub bileşen tanımlarını içermektedir. Ayrıca uygulamada kullanılacak Kafka’yı ayağa kaldırabilmek için bir docker-compose dosyasını da içerir.

docker-compose.yml

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
version: "3"
services:
  zookeeper:
    image: 'bitnami/zookeeper:3.8.0'
    ports:
      - '2181:2181'
    environment:
      - ALLOW_ANONYMOUS_LOGIN=yes
  kafka:
    image: 'bitnami/kafka:3.3.1'
    ports:
      - '9092:9092'
    environment:
      - KAFKA_BROKER_ID=1
      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092
      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://127.0.0.1:9092
      - KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper:2181
      - ALLOW_PLAINTEXT_LISTENER=yes
    depends_on:
      - zookeeper

  kafka-manager:
    image: hlebalbau/kafka-manager:stable
    ports:
      - "9000:9000"
    environment:
      ZK_HOSTS: zookeeper:2181
    restart: always

Docker.compose.yml dosyası, projede Kafka’yı kullanmak için gerekli olan verileri sağladığı için önemlidir.

redis-pubsub.yaml

1
2
3
4
5
6
7
8
9
10
11
12
apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: paymentpubsub
spec:
  type: pubsub.redis
  version: v1
  metadata:
  - name: redisHost
    value: localhost:6379
  - name: redisPassword
    value: ""

Redis-pubsub.yaml konfigürasyon dosyası, “paymentpubsub” adlı Dapr bileşenini temsil eder ve “paymentpubsub” bileşeni Redis üzerinde çalışan pub/sub sistemini kullanır.

• Metadata’da verilen isim uygulama içerisinde verilen isimle aynı olmalıdır.
• Name alanında verilen isim “paymentpubsub” uygulamanın hangi bileşenin kullanılacağını bilmesini sağlar.
• spec’in type’ı alanında yazılan state.redis bu bileşenin bir state mekanizması olduğunu ve Redis tipinde olduğunu tanımlamaktadır. Bu alanların alabileceği değerlere Dapr’ın sayfasından ulaşabilirsiniz.
• spec altında bulunan metadata kısmında ise bu state bileşeninin özelliklerinden bahsedilmektedir. Bu alanın alabileceği değerlere Dapr’ın ilgili sayfasından ulaşabilirsiniz.

StateStore.yaml

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: statestore
spec:
  type: state.redis
  version: v1
  metadata:
    - name: redisHost
      value: localhost:6379
    - name: redisPassword
      value: ""
    - name: outboxPublishPubsub # Required
      value: "paymentpubsub"
    - name: outboxPublishTopic # Required
      value: "paymentStatus"
  scopes:
    - payment

StateStore.yaml konfigürasyon dosyası, “statestore” adlı Dapr bileşenini temsil eder ve bu bileşen Redis tipindeki state store’u kullanmaktadır. Redis host’un altında tanımlanan diğer alanlar outbox deseninin aktif hale getirmek için kullanılmaktadır.

kafka-pubsub.yaml

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: paymentpubsub
spec:
  type: pubsub.kafka
  version: v1
  metadata:
    - name: brokers # Required. Kafka broker connection setting
      value: "localhost:9092"
    - name: authRequired
      value: "false"
    - name: maxMessageBytes # Optional.
      value: 1024
    - name: consumeRetryInterval # Optional.
      value: 200ms
    - name: disableTls # Optional. Disable TLS.
      value: "true"

Kafka-pubsub.yaml konfigürasyon dosyası, Dapr bileşeni olan “paymentpubsub” adlı pub/sub tipinde bir bileşeni temsil eder. Paymentpubsub bileşeni, Kafka mesajlaşma sistemini kullanmaktadır.

• metadata’da verilen isim uygulama içerisinde verilen isimle aynı olmalıdır.
• Name alanında verilen isim uygulamanın hangi bileşenin kullanılacağını bilmesini sağlar.
• spec’in type’ı alanında yazılan pubsub.kafka bu bileşenin bir pubsub mekanizması olduğunu ve Kafka tipinde olduğunu tanımlamaktadır. Bu alanların alabileceği değerlere Dapr’ın sayfasından ulaşabilirsiniz.
• spec altında bulunan metadata kısmında ise Kafka bileşeninin özelliklerinden bahsedilmektedir. Kafka 9092 portunda çalıştığı için burada brokers’ın adresine localhost:9092 tanımı yapılmaktadır. Bu alanın alabileceği değerlere Dapr’ın ilgili sayfasından ulaşabilirsiniz.

Uygulamanın mevcut durumunda, mesajlaşma alt yapısı olarak Redis kullanılmaktadır ve aynı zamanda veri tabanı için de Redis kullanılmaktadır. Kullandığımız mesajlaşma altyapısını Redis’in üzerinden Kafka’ya taşımak istediğimizde uygulama kodunda herhangi bir değişiklik yapmadan sadece konfigürasyon dosyaları değiştirmemiz yeterli olacaktır. Bunu sağlamak için kafka-pubsub.yaml dosyası kullanılacaktır.

• Payment klasörü içinde mvn clean install komutu çalıştırıldıktan sonra dapr run -f . komutunu çalıştırmak gerekmektedir. Bu işlem sonucunda, payment servisi başlatılacak ve Redis veri tabanı oluşturulacaktır. Ardından, mesajların Redis’te bulunan topic’e iletildiğini görüntüleyebiliriz. Bu aşamanın başarılı olduğunu gördükten sonra, uygulamanın mesajlaşma altyapısının Redis’ten Kafka’ya dönüşümünü gerçekleştirebiliriz.

• resources dizini içerisinde yer alan kafka-pubsub.yaml içeriği, payment içerisindeki pubsub.yaml dosyasına kopyalandığında Kafka’ya dönüşüm işlemini gerçekleştirmiş oluruz. İşlem sonrasında kullandığımız ortama göre pubsub.yaml dosyasını kaydetmemiz gerekmektedir.
• Uygulamayı yeniden çalıştırmadan önce Kafka’nın ayakta olduğunun kontrol edilmesi gerekmektedir. Eğer ayakta değilse resources dizini altında, docker-compose up -d komutu çalıştırılmalıdır. Bu komut docker-compose dosyasında tanımlı olan kafka konteynır’ını oluşturur.
• Kafka ayağa kalktıktan sonra dapr run -f . komutu ile payment çalıştırılır. Daha sonrasında dinleyen servisimizi çalıştırmak gerekmektedir. Bunun içinde shipment içerisinde mvn clean install yaptıktan sonra dapr run -f . komutu çalıştırılmalıdır.
• Bunun sonucunda dönüştürme işlemi başarı ile gerçekleşmiş olacaktır.

• Shipment servisi içerisindeki pubsub-yaml dosyası Kafka’ya göre düzenlendiği zaman başarılı bir şekilde topic’i dinlediğini ve mesajlara ulaştığını gözlemleriz.

Yazımızda görüldüğü üzere dönüşüm esnasında kaynak kodunda herhangi bir değişiklik yapmamız gerekmedi. Sadece kullanılan konfigürasyon dosyalarını değiştirerek mesaj alışverişinde kullanılan alt yapı değiştirilmiş oldu. Dapr sayesinde kod alt yapısal bağımlılıklardan tamamen soyutlanıp sadece iş kurallarına odaklı şekilde yazılabilir hale gelebildi. Dapr’ın birçok bileşen desteği sayesinde uygulamamızı çalıştıracağımız ortamların gereksinimlerine göre ilgili bileşenlerin konfigürasyonlarını tanımlayıp uygulamamızı hiç sorun olmadan çalıştırabiliriz.

Dapr’ın outbox özelliği, mikroservis mimarilerindeki uygulamaların işlem sırasını ve durumunu yönetmeyi amaçlar. Bu özellik sayesinde, gönderilen mesajlar geçici bir depolama alanında bekletilir ve alıcıya başarılı bir şekilde iletilip iletilmediği takip edilir. Outbox, asenkron iletişimi destekler ve özellikle dağıtık sistemlerdeki mikroservisler arasında güvenilir ve sıralı mesaj iletimini kolaylaştırır. Böylece uygulamaların daha tutarlı ve güvenilir bir şekilde çalışmasına olanak tanır. Dapr’ın outbox özelliği, dağıtık uygulama geliştiricilerine iletişim ve iş akışları konusunda daha güçlü bir kontrol sağlar.

Kaynakça

• https://dapr.io/