ibessonov commented on code in PR #7064: URL: https://github.com/apache/ignite-3/pull/7064#discussion_r2559134503
########## modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildTaskStatisticsLoggingListener.java: ########## @@ -0,0 +1,180 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.index; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import org.apache.ignite.internal.logger.IgniteLogger; +import org.apache.ignite.internal.logger.Loggers; +import org.apache.ignite.internal.tx.TxState; +import org.jetbrains.annotations.Nullable; +import org.jetbrains.annotations.TestOnly; + +/** + * Listener that collects {@link IndexBuildTask} statistics during execution and logs the aggregated results when the index build + * completes. + */ +class IndexBuildTaskStatisticsLoggingListener implements IndexBuildTaskListener { + private static final IgniteLogger LOG = Loggers.forClass(IndexBuildTaskStatisticsLoggingListener.class); + + private final IndexBuildTaskId taskId; + + private final boolean afterDisasterRecovery; + + private final AtomicLong startTime = new AtomicLong(); + + private final AtomicInteger successfulRaftCallCount = new AtomicInteger(0); + + private final AtomicInteger failedRaftCallCount = new AtomicInteger(0); + + private final AtomicLong rowIndexedCount = new AtomicLong(0); + + private final ConcurrentMap<TxState, AtomicInteger> resolvedWriteIntentCount = new ConcurrentHashMap<>(); + + IndexBuildTaskStatisticsLoggingListener(IndexBuildTaskId taskId, boolean afterDisasterRecovery) { + this.taskId = taskId; + this.afterDisasterRecovery = afterDisasterRecovery; + } + + @Override + public void onIndexBuildStarted(IndexBuildTaskId taskId) { + checkTaskId(taskId); + + startTime.set(System.currentTimeMillis()); + } + + @Override + public void onWriteIntentResolved(IndexBuildTaskId taskId, TxState txState) { + checkTaskId(taskId); + + resolvedWriteIntentCount.computeIfAbsent(txState, unused -> new AtomicInteger(0)).incrementAndGet(); + } + + @Override + public void onRaftCallSuccess(IndexBuildTaskId taskId) { + checkTaskId(taskId); + + successfulRaftCallCount.incrementAndGet(); + } + + @Override + public void onRaftCallFailure(IndexBuildTaskId taskId) { + checkTaskId(taskId); + + failedRaftCallCount.incrementAndGet(); + } + + @Override + public void onBatchProcessed(IndexBuildTaskId taskId, int rowCount) { + checkTaskId(taskId); + + rowIndexedCount.addAndGet(rowCount); + } + + @Override + public void onIndexBuildSuccess(IndexBuildTaskId taskId) { + checkTaskId(taskId); + + logStatistics(null); + } + + @Override + public void onIndexBuildFailure(IndexBuildTaskId taskId, Throwable throwable) { + checkTaskId(taskId); + + logStatistics(throwable); + } + + private void checkTaskId(IndexBuildTaskId taskId) { + if (!this.taskId.equals(taskId)) { + String message = String.format( + "Listener invoked with unexpected index id. Expected: [%s], but got: [%s]", + this.taskId, taskId + ); + + LOG.error(message); + throw new IllegalArgumentException(message); + } + } + + private void logStatistics(@Nullable Throwable throwable) { + String status = throwable == null + ? "success" + : String.format("failure (%s: %s)", throwable.getClass().getName(), throwable.getMessage()); + String reason = afterDisasterRecovery ? "disaster recovery of an AVAILABLE index" : "normal build"; + + LOG.info( + "Index build statistics: [" + + "task id: {}, " + + "status: {}, " + + "build reason: {}, " + + "time: {} ms, " + + "rows indexed: {}, " + + "successful raft calls: {}, " + + "failed raft calls: {}, " + + "resolved write intents: {}]", + taskId, + status, + reason, + getBuildTime(), + rowIndexedCount, + successfulRaftCallCount, + failedRaftCallCount, + resolvedWriteIntentCount + ); + } + + private long getBuildTime() { + if (startTime.get() == 0) { + String message = "Index build start time has not been set."; + LOG.error(message); + + throw new IllegalStateException(message); + } Review Comment: Should be an assertion instead of a runtime check I believe. Because this state is impossible ########## modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildTaskStatisticsLoggingListener.java: ########## @@ -0,0 +1,180 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.index; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import org.apache.ignite.internal.logger.IgniteLogger; +import org.apache.ignite.internal.logger.Loggers; +import org.apache.ignite.internal.tx.TxState; +import org.jetbrains.annotations.Nullable; +import org.jetbrains.annotations.TestOnly; + +/** + * Listener that collects {@link IndexBuildTask} statistics during execution and logs the aggregated results when the index build + * completes. + */ +class IndexBuildTaskStatisticsLoggingListener implements IndexBuildTaskListener { + private static final IgniteLogger LOG = Loggers.forClass(IndexBuildTaskStatisticsLoggingListener.class); + + private final IndexBuildTaskId taskId; + + private final boolean afterDisasterRecovery; + + private final AtomicLong startTime = new AtomicLong(); + + private final AtomicInteger successfulRaftCallCount = new AtomicInteger(0); + + private final AtomicInteger failedRaftCallCount = new AtomicInteger(0); + + private final AtomicLong rowIndexedCount = new AtomicLong(0); + + private final ConcurrentMap<TxState, AtomicInteger> resolvedWriteIntentCount = new ConcurrentHashMap<>(); + + IndexBuildTaskStatisticsLoggingListener(IndexBuildTaskId taskId, boolean afterDisasterRecovery) { + this.taskId = taskId; + this.afterDisasterRecovery = afterDisasterRecovery; + } + + @Override + public void onIndexBuildStarted(IndexBuildTaskId taskId) { + checkTaskId(taskId); + + startTime.set(System.currentTimeMillis()); + } + + @Override + public void onWriteIntentResolved(IndexBuildTaskId taskId, TxState txState) { + checkTaskId(taskId); + + resolvedWriteIntentCount.computeIfAbsent(txState, unused -> new AtomicInteger(0)).incrementAndGet(); + } + + @Override + public void onRaftCallSuccess(IndexBuildTaskId taskId) { + checkTaskId(taskId); + + successfulRaftCallCount.incrementAndGet(); + } + + @Override + public void onRaftCallFailure(IndexBuildTaskId taskId) { + checkTaskId(taskId); + + failedRaftCallCount.incrementAndGet(); + } + + @Override + public void onBatchProcessed(IndexBuildTaskId taskId, int rowCount) { + checkTaskId(taskId); + + rowIndexedCount.addAndGet(rowCount); + } + + @Override + public void onIndexBuildSuccess(IndexBuildTaskId taskId) { + checkTaskId(taskId); + + logStatistics(null); + } + + @Override + public void onIndexBuildFailure(IndexBuildTaskId taskId, Throwable throwable) { + checkTaskId(taskId); + + logStatistics(throwable); + } + + private void checkTaskId(IndexBuildTaskId taskId) { + if (!this.taskId.equals(taskId)) { + String message = String.format( + "Listener invoked with unexpected index id. Expected: [%s], but got: [%s]", + this.taskId, taskId + ); + + LOG.error(message); + throw new IllegalArgumentException(message); + } + } + + private void logStatistics(@Nullable Throwable throwable) { + String status = throwable == null + ? "success" + : String.format("failure (%s: %s)", throwable.getClass().getName(), throwable.getMessage()); Review Comment: I'd put that into a header of the log. Not sure what to do with the exception. Do you have an example of how this log will actually look? ########## modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildTaskStatisticsLoggingListener.java: ########## @@ -0,0 +1,180 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.index; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import org.apache.ignite.internal.logger.IgniteLogger; +import org.apache.ignite.internal.logger.Loggers; +import org.apache.ignite.internal.tx.TxState; +import org.jetbrains.annotations.Nullable; +import org.jetbrains.annotations.TestOnly; + +/** + * Listener that collects {@link IndexBuildTask} statistics during execution and logs the aggregated results when the index build + * completes. + */ +class IndexBuildTaskStatisticsLoggingListener implements IndexBuildTaskListener { + private static final IgniteLogger LOG = Loggers.forClass(IndexBuildTaskStatisticsLoggingListener.class); + + private final IndexBuildTaskId taskId; + + private final boolean afterDisasterRecovery; + + private final AtomicLong startTime = new AtomicLong(); + + private final AtomicInteger successfulRaftCallCount = new AtomicInteger(0); + + private final AtomicInteger failedRaftCallCount = new AtomicInteger(0); + + private final AtomicLong rowIndexedCount = new AtomicLong(0); + + private final ConcurrentMap<TxState, AtomicInteger> resolvedWriteIntentCount = new ConcurrentHashMap<>(); + + IndexBuildTaskStatisticsLoggingListener(IndexBuildTaskId taskId, boolean afterDisasterRecovery) { + this.taskId = taskId; + this.afterDisasterRecovery = afterDisasterRecovery; + } + + @Override + public void onIndexBuildStarted(IndexBuildTaskId taskId) { + checkTaskId(taskId); + + startTime.set(System.currentTimeMillis()); + } + + @Override + public void onWriteIntentResolved(IndexBuildTaskId taskId, TxState txState) { + checkTaskId(taskId); + + resolvedWriteIntentCount.computeIfAbsent(txState, unused -> new AtomicInteger(0)).incrementAndGet(); + } + + @Override + public void onRaftCallSuccess(IndexBuildTaskId taskId) { + checkTaskId(taskId); + + successfulRaftCallCount.incrementAndGet(); + } + + @Override + public void onRaftCallFailure(IndexBuildTaskId taskId) { + checkTaskId(taskId); + + failedRaftCallCount.incrementAndGet(); + } + + @Override + public void onBatchProcessed(IndexBuildTaskId taskId, int rowCount) { + checkTaskId(taskId); + + rowIndexedCount.addAndGet(rowCount); + } + + @Override + public void onIndexBuildSuccess(IndexBuildTaskId taskId) { + checkTaskId(taskId); + + logStatistics(null); + } + + @Override + public void onIndexBuildFailure(IndexBuildTaskId taskId, Throwable throwable) { + checkTaskId(taskId); + + logStatistics(throwable); + } + + private void checkTaskId(IndexBuildTaskId taskId) { + if (!this.taskId.equals(taskId)) { + String message = String.format( + "Listener invoked with unexpected index id. Expected: [%s], but got: [%s]", + this.taskId, taskId + ); + + LOG.error(message); + throw new IllegalArgumentException(message); + } + } + + private void logStatistics(@Nullable Throwable throwable) { + String status = throwable == null + ? "success" + : String.format("failure (%s: %s)", throwable.getClass().getName(), throwable.getMessage()); + String reason = afterDisasterRecovery ? "disaster recovery of an AVAILABLE index" : "normal build"; + + LOG.info( + "Index build statistics: [" + + "task id: {}, " + + "status: {}, " + + "build reason: {}, " + + "time: {} ms, " + + "rows indexed: {}, " + + "successful raft calls: {}, " + + "failed raft calls: {}, " + + "resolved write intents: {}]", Review Comment: ```suggestion + "taskId={}, " + "status={}, " + "buildReason={}, " + "time={}ms, " + "rowsIndexed={}, " + "successfulRaftCalls={}, " + "failedRaftCalls={}, " + "resolvedWriteIntents={}]", ``` ########## modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildTaskStatisticsLoggingListener.java: ########## @@ -0,0 +1,180 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.index; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import org.apache.ignite.internal.logger.IgniteLogger; +import org.apache.ignite.internal.logger.Loggers; +import org.apache.ignite.internal.tx.TxState; +import org.jetbrains.annotations.Nullable; +import org.jetbrains.annotations.TestOnly; + +/** + * Listener that collects {@link IndexBuildTask} statistics during execution and logs the aggregated results when the index build + * completes. + */ +class IndexBuildTaskStatisticsLoggingListener implements IndexBuildTaskListener { + private static final IgniteLogger LOG = Loggers.forClass(IndexBuildTaskStatisticsLoggingListener.class); + + private final IndexBuildTaskId taskId; + + private final boolean afterDisasterRecovery; + + private final AtomicLong startTime = new AtomicLong(); + + private final AtomicInteger successfulRaftCallCount = new AtomicInteger(0); + + private final AtomicInteger failedRaftCallCount = new AtomicInteger(0); + + private final AtomicLong rowIndexedCount = new AtomicLong(0); + + private final ConcurrentMap<TxState, AtomicInteger> resolvedWriteIntentCount = new ConcurrentHashMap<>(); + + IndexBuildTaskStatisticsLoggingListener(IndexBuildTaskId taskId, boolean afterDisasterRecovery) { + this.taskId = taskId; + this.afterDisasterRecovery = afterDisasterRecovery; + } + + @Override + public void onIndexBuildStarted(IndexBuildTaskId taskId) { + checkTaskId(taskId); + + startTime.set(System.currentTimeMillis()); + } + + @Override + public void onWriteIntentResolved(IndexBuildTaskId taskId, TxState txState) { + checkTaskId(taskId); + + resolvedWriteIntentCount.computeIfAbsent(txState, unused -> new AtomicInteger(0)).incrementAndGet(); + } + + @Override + public void onRaftCallSuccess(IndexBuildTaskId taskId) { + checkTaskId(taskId); + + successfulRaftCallCount.incrementAndGet(); + } + + @Override + public void onRaftCallFailure(IndexBuildTaskId taskId) { + checkTaskId(taskId); + + failedRaftCallCount.incrementAndGet(); + } + + @Override + public void onBatchProcessed(IndexBuildTaskId taskId, int rowCount) { + checkTaskId(taskId); + + rowIndexedCount.addAndGet(rowCount); + } + + @Override + public void onIndexBuildSuccess(IndexBuildTaskId taskId) { + checkTaskId(taskId); + + logStatistics(null); + } + + @Override + public void onIndexBuildFailure(IndexBuildTaskId taskId, Throwable throwable) { + checkTaskId(taskId); + + logStatistics(throwable); + } + + private void checkTaskId(IndexBuildTaskId taskId) { + if (!this.taskId.equals(taskId)) { + String message = String.format( + "Listener invoked with unexpected index id. Expected: [%s], but got: [%s]", + this.taskId, taskId + ); + + LOG.error(message); + throw new IllegalArgumentException(message); + } + } + + private void logStatistics(@Nullable Throwable throwable) { + String status = throwable == null + ? "success" + : String.format("failure (%s: %s)", throwable.getClass().getName(), throwable.getMessage()); + String reason = afterDisasterRecovery ? "disaster recovery of an AVAILABLE index" : "normal build"; Review Comment: Let's simplify it and use `DISASTER_RECOVERY` / `BUILD` instead of full sentences. The reason why I'm asking for that is that `[foo=full sentence, bar=other sentence]` looks weird in logs, spaces distract you from properly parsing the message. When it comes to `=` sign instead of `:`, please take a look at `S.toString` usages. By the say, you can use it to avoid manual concatenations / formatting, it's pretty convenient for large log messages. ########## modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildTaskStatisticsLoggingListener.java: ########## @@ -0,0 +1,180 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.index; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import org.apache.ignite.internal.logger.IgniteLogger; +import org.apache.ignite.internal.logger.Loggers; +import org.apache.ignite.internal.tx.TxState; +import org.jetbrains.annotations.Nullable; +import org.jetbrains.annotations.TestOnly; + +/** + * Listener that collects {@link IndexBuildTask} statistics during execution and logs the aggregated results when the index build + * completes. + */ +class IndexBuildTaskStatisticsLoggingListener implements IndexBuildTaskListener { + private static final IgniteLogger LOG = Loggers.forClass(IndexBuildTaskStatisticsLoggingListener.class); + + private final IndexBuildTaskId taskId; + + private final boolean afterDisasterRecovery; + + private final AtomicLong startTime = new AtomicLong(); + + private final AtomicInteger successfulRaftCallCount = new AtomicInteger(0); + + private final AtomicInteger failedRaftCallCount = new AtomicInteger(0); + + private final AtomicLong rowIndexedCount = new AtomicLong(0); Review Comment: All these fields could just be `volatile` primitives. Can you please explain this choice? Maybe I'm missing something ########## modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildTask.java: ########## @@ -156,24 +160,34 @@ void start() { return; } - LOG.info("Start building the index: [{}]", createCommonIndexInfo()); + if (afterDisasterRecovery) { + LOG.warn("Start building the index: [{}] due to disaster recovery of an AVAILABLE index. This shouldn't normally occur", + createCommonIndexInfo() + ); Review Comment: The "normal" pattern for logs is when you have a message first, and then all additional properties in `[]` at the end. There's no consensus on `.` / `:` / ` ` usage in the separator. As far as I can tell, the majority of logs look like this: ``` Something happened [key=value, key=value] ``` Please reformat this message to follow this convention, thank you! We should convert all logs to a single standard eventually, I hope it'll happen one day. ########## modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildTaskListener.java: ########## @@ -0,0 +1,40 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.index; + +import org.apache.ignite.internal.tx.TxState; + +/** + * Listener for {@link IndexBuildTask}. + */ +interface IndexBuildTaskListener { Review Comment: Was it necessary to split the listener into an interface and implementation? In other words, are we expecting more than a single implementation? ########## modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildTask.java: ########## @@ -301,7 +324,12 @@ private CompletableFuture<TxState> resolveFinalTxStateIfNeeded(UUID transactionI ZonePartitionId commitGroupId = new ZonePartitionId(commitPartitionId.commitZoneId, commitPartitionId.commitPartitionId); - return finalTransactionStateResolver.resolveFinalTxState(transactionId, commitGroupId); + return finalTransactionStateResolver.resolveFinalTxState(transactionId, commitGroupId) + .thenApply(txState -> { + taskListener.onWriteIntentResolved(taskId, txState); + + return txState; Review Comment: You're free to modify the API so that `onWriteIntentResolved` returns the `txState`, and then simplify this code. And if you get rid of `taskId` then it'll become ``` .thenApply(taskListener::onWriteIntentResolved) ``` which is the most desired option in my opinion. ########## modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildTask.java: ########## @@ -125,7 +127,8 @@ class IndexBuildTask { IgniteSpinBusyLock busyLock, int batchSize, InternalClusterNode node, - List<IndexBuildCompletionListener> listeners, + List<IndexBuildCompletionListener> buildCompletionListeners, + IndexBuildTaskListener taskListener, Review Comment: Could you please explain why it is a constructor parameter instead of just instantiating a field inside of the constructor? Is this for a future use in metrics? ########## modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildTaskStatisticsLoggingListener.java: ########## @@ -0,0 +1,180 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.index; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import org.apache.ignite.internal.logger.IgniteLogger; +import org.apache.ignite.internal.logger.Loggers; +import org.apache.ignite.internal.tx.TxState; +import org.jetbrains.annotations.Nullable; +import org.jetbrains.annotations.TestOnly; + +/** + * Listener that collects {@link IndexBuildTask} statistics during execution and logs the aggregated results when the index build + * completes. + */ +class IndexBuildTaskStatisticsLoggingListener implements IndexBuildTaskListener { + private static final IgniteLogger LOG = Loggers.forClass(IndexBuildTaskStatisticsLoggingListener.class); + + private final IndexBuildTaskId taskId; + + private final boolean afterDisasterRecovery; + + private final AtomicLong startTime = new AtomicLong(); + + private final AtomicInteger successfulRaftCallCount = new AtomicInteger(0); + + private final AtomicInteger failedRaftCallCount = new AtomicInteger(0); + + private final AtomicLong rowIndexedCount = new AtomicLong(0); + + private final ConcurrentMap<TxState, AtomicInteger> resolvedWriteIntentCount = new ConcurrentHashMap<>(); + + IndexBuildTaskStatisticsLoggingListener(IndexBuildTaskId taskId, boolean afterDisasterRecovery) { + this.taskId = taskId; + this.afterDisasterRecovery = afterDisasterRecovery; + } + + @Override + public void onIndexBuildStarted(IndexBuildTaskId taskId) { + checkTaskId(taskId); + + startTime.set(System.currentTimeMillis()); Review Comment: Usually we use `System.nanoTime()` for measuring time, it is more precise and it's guaranteed to be monotonous. ########## modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildTask.java: ########## @@ -221,9 +235,18 @@ private CompletableFuture<Void> handleNextBatch(@Nullable RowId highestRowId) { try { return createBatchToIndex(highestRowId) - .thenCompose(batch -> { - return replicaService.invoke(node, createBuildIndexReplicaRequest(batch, initialOperationTimestamp)); - }) + .thenCompose(batch -> + replicaService.invoke(node, createBuildIndexReplicaRequest(batch, initialOperationTimestamp)) + .whenComplete((unused, throwable) -> { + if (throwable == null) { + taskListener.onRaftCallSuccess(taskId); + } else { + taskListener.onRaftCallFailure(taskId); + } + }) + .thenApply(unused -> batch) + ) + .thenAccept(batch -> taskListener.onBatchProcessed(taskId, batch.rowIds.size())) Review Comment: Here I see a multiple approaches mixed up. It makes the code hard to read. For example, why is it ``` thenCompose( whenComplete(...) ) .thenAccept ``` instead of ``` thenCompose(...) .whenComplete(...) .thenAccept ``` ? The more you nest the code, the more complicated it becomes. Please consider chaining your steps on a higher level, and maybe extracting some of them into separate methods to make lambda expressions smaller. Thank you! ########## modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildTask.java: ########## @@ -156,24 +160,34 @@ void start() { return; } - LOG.info("Start building the index: [{}]", createCommonIndexInfo()); + if (afterDisasterRecovery) { + LOG.warn("Start building the index: [{}] due to disaster recovery of an AVAILABLE index. This shouldn't normally occur", + createCommonIndexInfo() + ); + } else { + LOG.info("Start building the index: [{}]", createCommonIndexInfo()); + } try { + taskListener.onIndexBuildStarted(taskId); + supplyAsync(partitionStorage::highestRowId, executor) .thenApplyAsync(this::handleNextBatch, executor) .thenCompose(Function.identity()) .whenComplete((unused, throwable) -> { if (throwable != null) { + String errorMessage = String.format("Index build error: [%s]", createCommonIndexInfo()); if (ignorable(throwable)) { - LOG.debug("Index build error: [{}]", throwable, createCommonIndexInfo()); + LOG.info(errorMessage, throwable); Review Comment: I'd prefer having separate messages. Currently you don't include any details about the `ignorable` error, which is one of the things that I asked to provide in issue's description. Please change this code accordingly, thank you! ########## modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildTaskStatisticsLoggingListener.java: ########## @@ -0,0 +1,180 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.index; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import org.apache.ignite.internal.logger.IgniteLogger; +import org.apache.ignite.internal.logger.Loggers; +import org.apache.ignite.internal.tx.TxState; +import org.jetbrains.annotations.Nullable; +import org.jetbrains.annotations.TestOnly; + +/** + * Listener that collects {@link IndexBuildTask} statistics during execution and logs the aggregated results when the index build + * completes. + */ +class IndexBuildTaskStatisticsLoggingListener implements IndexBuildTaskListener { + private static final IgniteLogger LOG = Loggers.forClass(IndexBuildTaskStatisticsLoggingListener.class); + + private final IndexBuildTaskId taskId; + + private final boolean afterDisasterRecovery; + + private final AtomicLong startTime = new AtomicLong(); + + private final AtomicInteger successfulRaftCallCount = new AtomicInteger(0); + + private final AtomicInteger failedRaftCallCount = new AtomicInteger(0); + + private final AtomicLong rowIndexedCount = new AtomicLong(0); + + private final ConcurrentMap<TxState, AtomicInteger> resolvedWriteIntentCount = new ConcurrentHashMap<>(); + + IndexBuildTaskStatisticsLoggingListener(IndexBuildTaskId taskId, boolean afterDisasterRecovery) { + this.taskId = taskId; + this.afterDisasterRecovery = afterDisasterRecovery; + } + + @Override + public void onIndexBuildStarted(IndexBuildTaskId taskId) { + checkTaskId(taskId); + + startTime.set(System.currentTimeMillis()); + } + + @Override + public void onWriteIntentResolved(IndexBuildTaskId taskId, TxState txState) { + checkTaskId(taskId); + + resolvedWriteIntentCount.computeIfAbsent(txState, unused -> new AtomicInteger(0)).incrementAndGet(); + } + + @Override + public void onRaftCallSuccess(IndexBuildTaskId taskId) { + checkTaskId(taskId); + + successfulRaftCallCount.incrementAndGet(); + } + + @Override + public void onRaftCallFailure(IndexBuildTaskId taskId) { + checkTaskId(taskId); + + failedRaftCallCount.incrementAndGet(); + } + + @Override + public void onBatchProcessed(IndexBuildTaskId taskId, int rowCount) { + checkTaskId(taskId); + + rowIndexedCount.addAndGet(rowCount); + } + + @Override + public void onIndexBuildSuccess(IndexBuildTaskId taskId) { + checkTaskId(taskId); + + logStatistics(null); + } + + @Override + public void onIndexBuildFailure(IndexBuildTaskId taskId, Throwable throwable) { + checkTaskId(taskId); + + logStatistics(throwable); + } + + private void checkTaskId(IndexBuildTaskId taskId) { Review Comment: Is this really required? How can there be any risk of having a wrong ID here? -- 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]
