Implementation of replica syncing monitor #119
Conversation
ReplicaSyncingMonitor#dateRate shouldn't be a implementation detail
Use --interval to specify the frequency of replics progress check. This is a useful flag for user who want faster feedback, also the existence of this flag enable us to have the tests run quicker.
chia7712
left a comment
chia7712
left a comment
There was a problem hiding this comment.
@garyparrot 能否麻煩更新readme?
| @CsvSource( | ||
| delimiterString = ",", | ||
| value = { | ||
| // leaderSize, previousSize, currentSize, interval, dataRatePerSec, Progress, Remaining, ? |
There was a problem hiding this comment.
最後那個多餘的欄位是在描述這筆測試輸入的意義，當初好像因為寫太長會讓 spotless 嘗試將 comment 分到下一行，所以沒辦法取一些比較有意義的名稱(比如 test-purpose)
我可以試著把他弄得比較正常一些
app/src/test/java/org/astraea/topic/ReplicaSyncingMonitorTest.java
Outdated
Show resolved
Hide resolved
| .collect(Collectors.joining(" ", "", " estimated")); | ||
| } | ||
|
|
||
| public String dataRateString() { |
There was a problem hiding this comment.
這個動作在其他工具應該也會用到，看看能否寫成一個公用的class讓大家可以輕鬆將數字在不同單位轉換
There was a problem hiding this comment.
如果你同意我上述的意見的話，麻煩先開一個議題，然後將議題編號回覆到此討論串
There was a problem hiding this comment.
@chia7712 我應該在這個 PR 內處理#121議題嗎，或是之後再開一個獨立的 PR 處理
this test will not report an error even if IllegalStateException is thrown. Since the exception is thrown on another Thread. The thread JUnit is guarding won't feel anything. This fix now run the execution on the same thread to avoid this situation.
chia7712
left a comment
chia7712
left a comment
There was a problem hiding this comment.
@garyparrot 感謝回覆，還有一些意見請看一下
|
|
||
| static void execute(final TopicAdmin topicAdmin, final Argument argument) { | ||
|
|
||
| // this supplier will gives you all the topic name that the client interest in. |
| converter = MillisecondConverter.class) | ||
| public int interval = 1000; | ||
|
|
||
| public static class MillisecondConverter implements IStringConverter<Integer> { |
There was a problem hiding this comment.
喔 我本來的意思是說 可以的話盡量共用時間的轉換 (see ArgumentUtil#DurationConverter)
There was a problem hiding this comment.
我剛剛看了一下 ArgumentUtil#DurationConverter 似乎只能做到 Second 單位的讀取，可能不適合這種可以接受 millisecond 的。
There was a problem hiding this comment.
只能用到秒單位對某些人來說可能有點慢，可能有人想 0.5 秒(500ms)看一次進度。
https://github.com/skiptests/astraea/pull/119/files#diff-5207e4edbe5dc490ae59e46c341451f51ebecb8829f9bd4cdca14c1b82858367R59
然後測試裡面有一些情境需要從 main 執行來測試工具有沒有滿足某些業務需求。如果裡面一秒跑一輪的話可能整個測試會跑很久，像是上面那個連接就使用 10ms 來加快這個測試的速度
There was a problem hiding this comment.
另一個角度是說我們應該要支持“有單位“這件事情，例如可以輸入 10s or 10ms 這類的數字來作為“時間區段"
There was a problem hiding this comment.
那我再把它修改成能夠支援數字和時間單位的形式，可能寫一個專門給時間的 Converter
There was a problem hiding this comment.
這可能要在這隻PR完成比較好 這樣可以減少參數改動的次數
| public Set<String> topics = Set.of(EVERY_TOPIC); | ||
|
|
||
| @Parameter( | ||
| names = {"--keep-track"}, |
| static void execute(final TopicAdmin topicAdmin, final Argument argument) { | ||
|
|
||
| // this supplier will gives you all the topic name that the client interest in. | ||
| Supplier<Set<String>> topicToTrack = |
There was a problem hiding this comment.
這邊好像沒辦法 var, 編譯器沒辦法 infer 他的 type
/home/garyparrot/Programming/astraea/app/src/main/java/org/astraea/topic/ReplicaSyncingMonitor.java:33: error: cannot infer type for local variable topicToTrack
var topicToTrack =
^
(lambda expression needs an explicit target-type)
There was a problem hiding this comment.
等下，為何這邊要用Supplier? 此時此刻就先決定有哪些topics不行嗎？
There was a problem hiding this comment.
這和我當初定義的 --topic 的語義有關，--topic 留空時，會選擇叢集內的 non-synced replica 做監控
但是 --track flag 有一個需求是除了能持續執行，還要能挖掘其他新出現的 non-synced replica。在程式開始執行時就固定會違反 --track 的需求。
當初會弄成這樣是因為我覺得這個情境應該蠻常見的，可能某人想持續監控任何新的 non-synced replica, 而非要經常重跑程式刷新。
| names = {"--topic"}, | ||
| description = "String: topics to track", | ||
| validateWith = ArgumentUtil.NotEmptyString.class) | ||
| public Set<String> topics = Set.of(EVERY_TOPIC); |
There was a problem hiding this comment.
要否直接用empty set來代表所有topics?
| .collect(Collectors.joining(" ", "", " estimated")); | ||
| } | ||
|
|
||
| public String dataRateString() { |
There was a problem hiding this comment.
如果你同意我上述的意見的話，麻煩先開一個議題，然後將議題編號回覆到此討論串
Due to some complicate issue the default value of --interval looks like 1000 in help message even if it means 1 second instead of 1000 seconds. This commit fixes this by changing the type of --interval argument to double. This commit also enhances the test by covering the illegal interval value.
This reverts commit f6c06f7. Instead of deciding which time unit for --interval flag to use. We can just let users make this decision. In the next commit, I will implement a converter that lets users specify their time unit.
Now flag --interval support time unit with the help of TimeConverter, following string are valid argument for --interval. * 30days * 30day * 24h * 12m * 50s * 20 * 500ms * 500us * 500ns
chia7712
left a comment
chia7712
left a comment
There was a problem hiding this comment.
@garyparrot 感謝如此棒的patch，剩一些小建議，請看一下，然後就來合併！
README.md
Outdated
| 5. [Kafka metric client](#kafka-metric-client): utility for accessing kafka Mbean metrics via JMX. | ||
| 6. [Replica Collie](#replica-collie): move replicas from brokers to others. You can use this tool to obstruct specific brokers from hosting specific topics. | ||
| 7. [Kafka partition score](#Kafka-partition-score): score all broker's partitions. | ||
| 8. [Kafka relica syncing monitor](#Kafka-replica-syncing-monitor): Tracking replica syncing progress. |
| * <li><b>(doesn't work)</b> {@code "0.5" to {@code Duration.ofMillis(500)}} | ||
| * </ul> | ||
| */ | ||
| public static class TimeConverter implements IStringConverter<Duration>, IParameterValidator { |
There was a problem hiding this comment.
可否改名為DurationConverter，並且把舊版的刪掉?
| static void execute(final TopicAdmin topicAdmin, final Argument argument) { | ||
|
|
||
| // this supplier will gives you all the topic name that the client interest in. | ||
| Supplier<Set<String>> topicToTrack = |
There was a problem hiding this comment.
等下，為何這邊要用Supplier? 此時此刻就先決定有哪些topics不行嗎？
| () -> argument.topics.isEmpty() ? topicAdmin.topicNames() : argument.topics; | ||
|
|
||
| // the non-synced topic-partition we want to monitor | ||
| Set<TopicPartition> topicPartitionToTrack = |
| findNonSyncedTopicPartition(topicAdmin, topicToTrack.get()); | ||
|
|
||
| // keep tracking the previous replica size of a topic-partition-replica tuple | ||
| final Map<TopicPartitionReplica, Long> previousCheckedSize = new HashMap<>(); |
| static class Argument extends BasicArgumentWithPropFile { | ||
|
|
||
| @Parameter( | ||
| names = {"--topic"}, |
| public void validate(String name, String value) throws ParameterException { | ||
| if (!timePattern.matcher(value).find()) | ||
| throw new ParameterException( | ||
| "field \"" + name + "\"'s value \"" + value + "\" doesn't match time format"); |
| .collect(Collectors.toUnmodifiableList()); | ||
|
|
||
| @Test | ||
| void execute() throws InterruptedException { |
There was a problem hiding this comment.
Please remove InterruptedException
|
|
||
| class ReplicaSyncingMonitorIntegrationTest extends RequireBrokerCluster { | ||
|
|
||
| private static final String topicName = |
There was a problem hiding this comment.
考量是global variable，請用大寫字母表示
| " 60h, true , valid unit", | ||
| " 365day, true , valid unit", | ||
| " 365days, true , valid unit", | ||
| " 0010100days, true , valid unit", |
There was a problem hiding this comment.
我在思考 0 開頭的數字算不算合格的輸入，我個人覺得這個應該不能過
There was a problem hiding this comment.
- Pattern.compile( "^(?<value>[0-9]+)(?<unit>days|day|h|m|s|ms|us|ns|)$")
+ Pattern.compile("^(?!0[0-9])(?<value>[0-9]+)(?<unit>days|day|h|m|s|ms|us|ns|)$")如果在前面套用上 negative lookahead 的話應該可行
不過我後來覺得算了，有 leading zero 的數字似乎要說他並非數字好像也不太對勁。
chia7712
left a comment
chia7712
left a comment
There was a problem hiding this comment.
@garyparrot 這隻patch很棒！只剩一個小style麻煩改一下，謝謝
| */ | ||
| public static class DurationConverter implements IStringConverter<Duration>, IParameterValidator { | ||
|
|
||
| static final Pattern timePattern = |
There was a problem hiding this comment.
private static final Pattern TIME_PATTERN
2 participants
resolve #89
This PR implements a tool to monitor replica syncing progress, it can be used to monitor replica migration progress.
Flags
Example Code
Demo
TODO
README.md