Support transactionId and multi records in POST /records #464

brandboat · 2022-07-09T11:33:40Z

resolve #432

chia7712

@brandboat 感謝實作此好用的APIs，幾個意見請看一下，謝謝

chia7712 · 2022-07-09T11:38:24Z

app/src/main/java/org/astraea/app/web/RecordHandler.java

+                    Producer.builder()
+                        .transactionId(transactionId)
+                        .bootstrapServers(bootstrapServers)
+                        .build())


麻煩請明確使用buildTransactional

chia7712 · 2022-07-09T11:39:18Z

docs/web_server/web_api_records_chinese.md

+| 名稱            | 說明                                        | 預設值   |
+|---------------|-------------------------------------------|-------|
+| records       | (必填) 本次寫入資料，需填寫至少一比資料                     | 無     |
+| transactionId | (選填) 填入 transaction id 以執行 transaction 寫入 | 無     |


請補充說這個值是字串

chia7712 · 2022-07-09T11:42:06Z

app/src/main/java/org/astraea/app/web/RecordHandler.java

+                    // this shouldn't happen, but still add error 404 here
+                    results.add(Response.for404("missing result"));
+                  }
+                  latch.countDown();


請使用try-finally以免出現意外導致永遠等不到latch被完成

chia7712 · 2022-07-09T11:43:51Z

app/src/main/java/org/astraea/app/web/RecordHandler.java

  static final String PARTITION = "partition";
  static final String ASYNC = "async";
  static final String DISTANCE_FROM_LATEST = "distanceFromLatest";
  static final String DISTANCE_FROM_BEGINNING = "distanceFromBeginning";
  static final String SEEK_TO = "seekTo";
  static final String KEY_DESERIALIZER = "keyDeserializer";
  static final String VALUE_DESERIALIZER = "valueDeserializer";
-  static final String RECORDS = "records";
+  static final String LIMIT = "limit";


這個名稱更換是為了避免跟post batch裡出現的records重複嗎?

是的，另外一個原因是我覺得 limit 的命名會更清楚一些

chia7712 · 2022-07-09T11:51:08Z

app/src/main/java/org/astraea/app/web/RecordHandler.java

+            .get(TRANSACTION_ID)
+            .map(
+                transactionId ->
+                    Producer.builder()


如果使用者在不斷送出交易行為時，這邊重複建立的成本會很高，可否開個議題來討論如何實作快取transactional producer? 並且在這裡留個TODO

chia7712 · 2022-07-09T11:51:46Z

app/src/main/java/org/astraea/app/web/RecordHandler.java

+    }
+
+    @Override
+    public boolean equals(Object obj) {


這個在production code有用到嗎？

沒有，僅在 test code 裡頭用以確認資料一致

抱歉，這是我之前加測試時用到的程式碼，但後來把該測試移除了，便不再需要，我會在下一個 commit 移除，感謝

chia7712 · 2022-07-09T11:52:15Z

app/src/main/java/org/astraea/app/web/RecordHandler.java

+        request
+            .get(RECORDS)
+            .map(toPostRecordList)
+            .filter(list -> list.size() > 0)


!list.isEmpty()

chia7712 · 2022-07-09T11:56:34Z

app/src/main/java/org/astraea/app/web/RecordHandler.java

+        (data) ->
+            new GsonBuilder()
+                .create()
+                .fromJson(data, new TypeToken<ArrayList<PostRecord>>() {}.getType());


這個邏輯可否整合到PostRequest裡面？讓其他handler也可以享用到

chia7712 · 2022-07-09T11:59:17Z

app/src/main/java/org/astraea/app/web/RecordHandler.java

+                  .collect(toList()));
+    } finally {
+      if (producer.transactional()) {
+        producer.close();


這邊會破壞async的邏輯，因為關閉producer的時候都會等到資料被處理完成

chia7712 · 2022-07-09T12:00:36Z

docs/web_server/web_api_records_chinese.md

+|---------------|-------------------------------------------|-------|
+| records       | (必填) 本次寫入資料，需填寫至少一比資料                     | 無     |
+| transactionId | (選填) 填入 transaction id 以執行 transaction 寫入 | 無     |
+| async         | (選填) 是否要等到資料寫入 topic 再回傳                  | false |


或許在寫入資料時應該要重複使用timeout這個概念，否則在sync+batch同時使用下，使用者很難知道是否要“繼續等下去"，你覺得呢?

好主意，我會再加上 timeout parameter

這個可以再開一隻議題處理

chia7712

@brandboat 感謝patch，功能大致上沒問題，不過有一些code style的地方需要討論，請看一下，謝謝

chia7712 · 2022-07-12T09:13:09Z

app/src/main/java/org/astraea/app/web/RecordHandler.java

+        CompletableFuture.supplyAsync(
+                () -> {
+                  try {
+                    return producer.send(


這個方法有些太長了，可否把這段放到獨立的method裡面去？

chia7712 · 2022-07-12T09:35:08Z

app/src/main/java/org/astraea/app/web/RecordHandler.java

+                })
+            .thenApply(
+                senderFutures -> {
+                  var latch = new CountDownLatch(senderFutures.size());


這一段又是async又是latch，這樣的風格不太好、閱讀起來也有點痛苦，可否考慮以下寫法：

新增一個方法來處理多的 CompletableFuture (仿照scala)，例如

static<T> CompletableFuture<List<T>> sequence(List<CompletableFuture<T>> com) { return CompletableFuture.allOf(com.toArray(new CompletableFuture<?>[0])) .thenApply(v -> com.stream() .map(CompletableFuture::join) .collect(Collectors.toList()) ); }

用上述方法來重寫這一段，例如：

.thenCompose( senderFutures -> sequence(senderFutures.stream().map(s -> s.handle((metadata, exception) -> { if (metadata != null) return new Metadata(metadata); if (exception != null) return Response.for500(exception.getMessage()); // this shouldn't happen, but still add error 404 here return Response.for404("missing result"); })).map(CompletionStage::toCompletableFuture).collect(Collectors.toUnmodifiableList())));

最後的輸出就可以改成：

if (async) return Response.ACCEPT; return Utils.packException(() -> new PostResponse(result.get(timeout.toNanos(), TimeUnit.NANOSECONDS)));

BTW, sequence這個方法麻煩放到Utils裡面，並且補上測試

chia7712 · 2022-07-12T09:50:33Z

app/src/main/java/org/astraea/app/web/RecordHandler.java

-    request.get(TIMESTAMP).ifPresent(t -> sender.timestamp(Long.parseLong(t)));
-    request.get(PARTITION).ifPresent(p -> sender.partition(Integer.parseInt(p)));
+    Optional<List<PostRecord>> recordsArg =
+        request.get(RECORDS, new TypeToken<ArrayList<PostRecord>>() {}.getType());


我原先的意思比較是說能夠做到只給class就好，每次都要寫出完整的轉換 (new TypeToken<ArrayList<PostRecord>>)有點囉唆，例如：

default <T> List<T> values(String key, Class<T> clz) { return new Gson().fromJson(value(key), TypeToken.getParameterized(ArrayList.class, clz).getType()); } default <T> T value(String key, Class<T> clz) { return new Gson().fromJson(value(key), TypeToken.get(clz).getType()); }

感謝，這樣更為簡潔。不過如果不使用回傳 Optional<...> 的形式，user 填寫的 json value 為 null 或者 ""，gson 在進行 fromJson 時會直接噴出 null pointer exception。不過我想這部分我們可以從文件中去強調某些 parameter 絕對不可為空來提醒使用者。 (update: 透過文件來規範，或者如果有詳細的檢核需求應在各自 handler 處理)

不過如果不使用回傳 Optional<...> 的形式，user 填寫的 json value 為 null 或者 ""，gson 在進行 fromJson 時會直接噴出 null pointer exception。不過我想這部分我們可以從文件中去強調某些 parameter 絕對不可為空來提醒使用者。

這個概念不錯，可以開一個議題來追蹤，理想上要能捕捉到是缺少哪個參數，並且將其包裝再回傳的錯誤訊息裡面，就像是value(String)會拋出NoSuchElementException這個例外，而這個例外會被包裝成404這個錯誤，並且在錯誤訊息裡會描述"the value for " + key + " is nonexistent"

chia7712

LGTM

chia7712 reviewed Jul 9, 2022

View reviewed changes

brandboat mentioned this pull request Jul 10, 2022

Find a way to cache transactional producer in RecordsHandler #473

Closed

brandboat force-pushed the feature/implement-transaction-post-records branch 2 times, most recently from 042f263 to 74722a6 Compare July 10, 2022 15:44

brandboat added 4 commits July 11, 2022 23:57

Rename records to limit

6a61bdf

Support transactionId and multiple records in POST /records

4298b60

Add doc

35463f2

Addressed chia7712's comment

210744f

brandboat force-pushed the feature/implement-transaction-post-records branch from 74722a6 to 210744f Compare July 11, 2022 15:57

chia7712 reviewed Jul 12, 2022

View reviewed changes

Address chai7712's comments

f0263fc

chia7712 approved these changes Jul 12, 2022

View reviewed changes

chia7712 merged commit 342934f into opensource4you:main Jul 12, 2022

brandboat deleted the feature/implement-transaction-post-records branch July 13, 2022 03:06

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Support transactionId and multi records in POST /records #464

Support transactionId and multi records in POST /records #464

brandboat commented Jul 9, 2022

chia7712 left a comment

chia7712 Jul 9, 2022

chia7712 Jul 9, 2022

chia7712 Jul 9, 2022

brandboat Jul 9, 2022

chia7712 Jul 9, 2022

brandboat Jul 9, 2022

chia7712 Jul 9, 2022

chia7712 Jul 9, 2022

brandboat Jul 9, 2022

brandboat Jul 10, 2022

chia7712 Jul 9, 2022

chia7712 Jul 9, 2022

chia7712 Jul 9, 2022

chia7712 Jul 9, 2022

brandboat Jul 10, 2022

chia7712 Jul 10, 2022

chia7712 left a comment

chia7712 Jul 12, 2022

chia7712 Jul 12, 2022

chia7712 Jul 12, 2022

chia7712 Jul 12, 2022

brandboat Jul 12, 2022 •

edited

Loading

chia7712 Jul 12, 2022

chia7712 left a comment

Support transactionId and multi records in POST /records #464

Support transactionId and multi records in POST /records #464

Conversation

brandboat commented Jul 9, 2022

chia7712 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chia7712 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

brandboat Jul 12, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chia7712 left a comment

Choose a reason for hiding this comment

brandboat Jul 12, 2022 •

edited

Loading