laskoviymishka commented on code in PR #17025:
URL: https://github.com/apache/iceberg/pull/17025#discussion_r3567167629
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Channel.java:
##########
@@ -125,11 +128,17 @@ record -> {
// so increment the record offset by one
controlTopicOffsets.put(record.partition(), record.offset() + 1);
+ long readStart = System.nanoTime();
Event event = AvroUtil.decode(record.value());
+ getChannelMetrics().recordMessageRead((System.nanoTime() -
readStart) / 1_000_000L);
Review Comment:
The per-record timers I asked about last round are exactly what I wanted
here — one snag though: `/ 1_000_000L` is integer division, so any decode
faster than a millisecond floors to 0, and Avro-decoding a single control
message almost always is. Same on `recordMessageProcess` at line 140.
So in practice these two read 0 nearly all the time. I'd record microseconds
(`/ 1_000L`) and tweak the description, or pass the raw nanos as a double so
avg/max stay meaningful.
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/Coordinator.java:
##########
@@ -123,6 +129,7 @@ void process() {
Event event =
new Event(config.connectGroupId(), new
StartCommit(commitState.currentCommitId()));
send(event);
+ coordinatorMetrics.incStartCommit();
Review Comment:
Small semantics question on the counter placement, in the same spirit as the
count discussion last round: `incStartCommit()` sits after `send()`, so if
`send()` throws the commit is already in progress but the counter never moves.
Same shape in `Worker.receive()` — `incDataWritten`/`incDataComplete` run after
`send()`, so a failed send means the files are already on disk but the counters
show nothing.
If we're strictly counting "events successfully emitted", after-send is
right and I'd just leave a note saying so. If we want "commits initiated" /
"files written", I'd move them ahead of `send()` or into a finally. Either's
defensible — I'd just pick one on purpose. wdyt?
##########
kafka-connect/kafka-connect/src/main/java/org/apache/iceberg/connect/channel/CommitState.java:
##########
@@ -102,6 +102,14 @@ void clearResponses() {
commitBuffer.clear();
}
+ int commitBufferSize() {
+ return commitBuffer.size();
Review Comment:
This is the one thing from last round I still don't see addressed — the
gauge suppliers read `commitBuffer.size()`/`readyBuffer.size()` from the JMX
poll thread, but the buffers are plain `ArrayList`s mutated on the coordinator
thread with no happens-before between the two.
Best case the gauge just lags a cycle, which is fine; worst case a `size()`
read races an `add()` array-grow or a `clear()` that nulls elements before
resetting the size field, which is a real (if rare) window. I'd back these with
an `AtomicInteger` in `CommitState` bumped alongside the add/clear calls, or
read them under the same lock the coordinator mutates under — either's fine, I
just don't want the cross-thread `ArrayList.size()`. wdyt?
##########
kafka-connect/kafka-connect/src/test/java/org/apache/iceberg/connect/channel/ChannelTestBase.java:
##########
@@ -102,6 +102,8 @@ public void before() {
when(config.commitThreads()).thenReturn(1);
when(config.connectGroupId()).thenReturn(CONNECT_CONSUMER_GROUP_ID);
when(config.tableConfig(any())).thenReturn(mock(TableSinkConfig.class));
+ when(config.connectorName()).thenReturn("test-connector");
+ when(config.taskId()).thenReturn("0");
Review Comment:
Now that `before()` pins a fixed connector/task and both the `Worker` and
`Coordinator` constructors register MBeans, any test that builds one off this
config and doesn't `stop()` it leaves that bean registered for the rest of the
JVM — the next construction with the same name throws
`InstanceAlreadyExistsException`, and JUnit gives us no ordering guarantee.
`coordinatorTest()` already got the right treatment with the
`terminate()`/`stop()` finally. `testSave` in `TestWorker` and
`testPartialCommitFailureMetric` in `TestCoordinator` still construct directly
without it — I'd wrap those the same way.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
