stevenzwu commented on code in PR #7360: URL: https://github.com/apache/iceberg/pull/7360#discussion_r1171934280
########## flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/GlobalStatisticsAggregator.java: ########## @@ -0,0 +1,83 @@ +/* + * 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.iceberg.flink.sink.shuffle; + +import java.io.Serializable; +import java.util.Set; +import org.apache.iceberg.relocated.com.google.common.base.MoreObjects; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; +import org.apache.iceberg.relocated.com.google.common.collect.Sets; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * GlobalStatisticsAggregator is used by {@link DataStatisticsCoordinator} to collect {@link + * DataStatistics} from {@link DataStatisticsOperator} subtasks for specific checkpoint. It stores + * the merged {@link DataStatistics} result and uses set to keep a record of all reported subtasks. + */ +class GlobalStatisticsAggregator<K> implements Serializable { + private static final Logger LOG = LoggerFactory.getLogger(GlobalStatisticsAggregator.class); + + private final long checkpointId; + private final DataStatistics<K> dataStatistics; + private final Set<Integer> subtaskSet = Sets.newHashSet(); + + GlobalStatisticsAggregator(long checkpoint, final DataStatisticsFactory<K> statisticsFactory) { + this.checkpointId = checkpoint; + this.dataStatistics = statisticsFactory.createDataStatistics(); + } + + long checkpointId() { + return checkpointId; + } + + DataStatistics<K> dataStatistics() { + return dataStatistics; + } + + void mergeDataStatistic(int subtask, DataStatisticsEvent<K> event) { + Preconditions.checkArgument( + checkpointId == event.checkpointId(), + "Received unexpected event from checkpoint %s. Expected checkpoint %s", + event.checkpointId(), + checkpointId); + if (!subtaskSet.add(subtask)) { + LOG.debug( + "Receive duplicated data statistics for checkpoint {} subtask {}. Ignore it.", Review Comment: nit: the error msg can be a little more readable `Ignore duplicate data statistics from subtask {} for checkpoint {}` ########## flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsCoordinator.java: ########## @@ -0,0 +1,321 @@ +/* + * 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.iceberg.flink.sink.shuffle; + +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import org.apache.flink.annotation.Internal; +import org.apache.flink.runtime.operators.coordination.OperatorCoordinator; +import org.apache.flink.runtime.operators.coordination.OperatorEvent; +import org.apache.flink.util.ExceptionUtils; +import org.apache.flink.util.InstantiationUtil; +import org.apache.flink.util.Preconditions; +import org.apache.flink.util.function.ThrowingRunnable; +import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting; +import org.jetbrains.annotations.Nullable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * DataStatisticsCoordinator receives {@link DataStatisticsEvent} from {@link + * DataStatisticsOperator} every subtask and then merge them together. Once aggregation for all + * subtasks data statistics completes, DataStatisticsCoordinator will send the aggregated + * result(global data statistics) back to {@link DataStatisticsOperator}. In the end a custom + * partitioner will distribute traffic based on the global data statistics to improve data + * clustering. + */ +@Internal +class DataStatisticsCoordinator<K> implements OperatorCoordinator { + private static final Logger LOG = LoggerFactory.getLogger(DataStatisticsCoordinator.class); + + private static final double EXPECTED_DATA_STATISTICS_RECEIVED_PERCENTAGE = 0.8; + + private final String operatorName; + // A single-thread executor to handle all the actions for coordinator + private final ExecutorService coordinatorExecutor; + private final DataStatisticsCoordinatorContext<K> context; + private final DataStatisticsFactory<K> statisticsFactory; + + private volatile GlobalStatisticsAggregator<K> inProgressAggregator; + private volatile GlobalStatisticsAggregator<K> lastCompletedAggregator; + private volatile boolean started; + + DataStatisticsCoordinator( + String operatorName, + OperatorCoordinator.Context context, + DataStatisticsFactory<K> statisticsFactory) { + this.operatorName = operatorName; + DataStatisticsCoordinatorProvider.CoordinatorExecutorThreadFactory coordinatorThreadFactory = + new DataStatisticsCoordinatorProvider.CoordinatorExecutorThreadFactory( + "DataStatisticsCoordinator-" + operatorName, context.getUserCodeClassloader()); + this.coordinatorExecutor = Executors.newSingleThreadExecutor(coordinatorThreadFactory); + this.context = + new DataStatisticsCoordinatorContext<>( + coordinatorExecutor, coordinatorThreadFactory, context); + this.statisticsFactory = statisticsFactory; + } + + @Override + public void start() throws Exception { + LOG.info("Starting data statistics coordinator for {}.", operatorName); + started = true; + } + + @Override + public void close() throws Exception { + LOG.info("Closing data statistics coordinator for {}.", operatorName); + if (started) { + coordinatorExecutor.shutdown(); + if (!coordinatorExecutor.awaitTermination(5, TimeUnit.SECONDS)) { + LOG.warn("Fail to shut down DataStatisticsCoordinator gracefully. Shutting down now"); + coordinatorExecutor.shutdownNow(); + } + } + } + + private void runInCoordinatorThread( + ThrowingRunnable<Throwable> action, String actionName, Object... actionNameFormatParameters) { + ensureStarted(); + coordinatorExecutor.execute( + () -> { + try { + action.run(); + } catch (Throwable t) { + // if we have a JVM critical error, promote it immediately, there is a good + // chance the logging or job failing will not succeed anymore + ExceptionUtils.rethrowIfFatalErrorOrOOM(t); + + String actionString = String.format(actionName, actionNameFormatParameters); + LOG.error( + "Uncaught exception in the data statistics {} while {}. Triggering job failover.", + operatorName, + actionString, + t); + context.failJob(t); + } + }); + } + + private void ensureStarted() { + if (!this.started) { Review Comment: nit: Iceberg uses `Preconditions` ########## flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/GlobalStatisticsAggregator.java: ########## @@ -0,0 +1,83 @@ +/* + * 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.iceberg.flink.sink.shuffle; + +import java.io.Serializable; +import java.util.Set; +import org.apache.iceberg.relocated.com.google.common.base.MoreObjects; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; +import org.apache.iceberg.relocated.com.google.common.collect.Sets; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * GlobalStatisticsAggregator is used by {@link DataStatisticsCoordinator} to collect {@link + * DataStatistics} from {@link DataStatisticsOperator} subtasks for specific checkpoint. It stores + * the merged {@link DataStatistics} result and uses set to keep a record of all reported subtasks. + */ +class GlobalStatisticsAggregator<K> implements Serializable { + private static final Logger LOG = LoggerFactory.getLogger(GlobalStatisticsAggregator.class); + + private final long checkpointId; + private final DataStatistics<K> dataStatistics; + private final Set<Integer> subtaskSet = Sets.newHashSet(); + + GlobalStatisticsAggregator(long checkpoint, final DataStatisticsFactory<K> statisticsFactory) { + this.checkpointId = checkpoint; + this.dataStatistics = statisticsFactory.createDataStatistics(); + } + + long checkpointId() { + return checkpointId; + } + + DataStatistics<K> dataStatistics() { + return dataStatistics; + } + + void mergeDataStatistic(int subtask, DataStatisticsEvent<K> event) { + Preconditions.checkArgument( + checkpointId == event.checkpointId(), + "Received unexpected event from checkpoint %s. Expected checkpoint %s", + event.checkpointId(), + checkpointId); + if (!subtaskSet.add(subtask)) { + LOG.debug( + "Receive duplicated data statistics for checkpoint {} subtask {}. Ignore it.", + checkpointId, + subtask); + return; + } + + dataStatistics.merge(event.dataStatistics()); + } + + long aggregatedSubtasksCount() { Review Comment: should return `int` ########## flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsCoordinator.java: ########## @@ -0,0 +1,324 @@ +/* + * 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.iceberg.flink.sink.shuffle; + +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeUnit; +import org.apache.flink.runtime.operators.coordination.OperatorCoordinator; +import org.apache.flink.runtime.operators.coordination.OperatorEvent; +import org.apache.flink.util.ExceptionUtils; +import org.apache.flink.util.FlinkException; +import org.apache.flink.util.InstantiationUtil; +import org.apache.flink.util.Preconditions; +import org.apache.flink.util.function.ThrowingRunnable; +import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting; +import org.jetbrains.annotations.Nullable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * DataStatisticsCoordinator receives {@link DataStatisticsEvent} from {@link + * DataStatisticsOperator} every subtask and then merge them together. Once aggregation for all + * subtasks data statistics completes, DataStatisticsCoordinator will send the aggregation + * result(global data statistics) back to {@link DataStatisticsOperator}. In the end a custom + * partitioner will distribute traffic based on the global data statistics to improve data + * clustering. + */ +class DataStatisticsCoordinator<K> implements OperatorCoordinator { + private static final Logger LOG = LoggerFactory.getLogger(DataStatisticsCoordinator.class); + + private static final double EXPECTED_DATA_STATISTICS_RECEIVED_PERCENTAGE = 0.8; + + private final String operatorName; + // A single-thread executor to handle all the actions for coordinator + private final ExecutorService coordinatorExecutor; + private final DataStatisticsCoordinatorContext<K> context; + private final DataStatisticsFactory<K> statisticsFactory; + + private volatile AggregateDataStatistics<K> incompleteAggregateDataStatistics; + private volatile AggregateDataStatistics<K> completeAggregateDataStatistics; + private volatile boolean started; + + public DataStatisticsCoordinator( + String operatorName, + ExecutorService coordinatorExecutor, + DataStatisticsCoordinatorContext<K> context, + DataStatisticsFactory<K> statisticsFactory) { + this.operatorName = operatorName; + this.coordinatorExecutor = coordinatorExecutor; + this.context = context; + this.statisticsFactory = statisticsFactory; + } + + @Override + public void start() throws Exception { + LOG.info("Starting data statistics coordinator for {}.", operatorName); + started = true; + } + + @Override + public void close() throws Exception { + LOG.info("Closing data statistics coordinator for {}.", operatorName); + try { + if (started) { + context.close(); + } + } finally { + coordinatorExecutor.shutdownNow(); + // We do not expect this to actually block for long. At this point, there should + // be very few task running in the executor, if any. + coordinatorExecutor.awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS); + } + } + + private void runInCoordinatorThread( + ThrowingRunnable<Throwable> action, String actionName, Object... actionNameFormatParameters) { + ensureStarted(); + coordinatorExecutor.execute( + () -> { + try { + action.run(); + } catch (Throwable t) { + // if we have a JVM critical error, promote it immediately, there is a good + // chance the logging or job failing will not succeed anymore + ExceptionUtils.rethrowIfFatalErrorOrOOM(t); + + final String actionString = String.format(actionName, actionNameFormatParameters); + LOG.error( + "Uncaught exception in the data statistics {} while {}. Triggering job failover.", + operatorName, + actionString, + t); + context.failJob(t); + } + }); + } + + private void ensureStarted() { + if (!this.started) { + throw new IllegalStateException("The coordinator has not started yet."); + } + } + + private void handleDataStatisticRequest(int subtask, DataStatisticsEvent<K> event) { + long checkpointId = event.checkpointId(); + + if (completeAggregateDataStatistics != null + && completeAggregateDataStatistics.checkpointId() >= checkpointId) { + LOG.debug( + "Data statistics aggregation for checkpoint {} has completed. Ignore the event from subtask {} for checkpoint {}", + completeAggregateDataStatistics.checkpointId(), + subtask, + checkpointId); + return; + } + + if (incompleteAggregateDataStatistics == null) { + incompleteAggregateDataStatistics = + new AggregateDataStatistics<>(checkpointId, statisticsFactory); + } + + if (incompleteAggregateDataStatistics.checkpointId() < checkpointId) { + if ((double) incompleteAggregateDataStatistics.aggregateSize() / context.currentParallelism() + >= EXPECTED_DATA_STATISTICS_RECEIVED_PERCENTAGE) { + completeAggregateDataStatistics = incompleteAggregateDataStatistics; + LOG.info( + "Received data statistics from {} operators out of total {} for checkpoint {}. It's more than the expected percentage {}. Thus sending the aggregate data statistics {} to subtasks.", + incompleteAggregateDataStatistics.aggregateSize(), + context.currentParallelism(), + incompleteAggregateDataStatistics.checkpointId(), + EXPECTED_DATA_STATISTICS_RECEIVED_PERCENTAGE, + completeAggregateDataStatistics); + incompleteAggregateDataStatistics = + new AggregateDataStatistics<>(checkpointId, statisticsFactory); + incompleteAggregateDataStatistics.mergeDataStatistic(subtask, event); + context.sendDataStatisticsToSubtasks( + incompleteAggregateDataStatistics.checkpointId(), + completeAggregateDataStatistics.dataStatistics()); + return; + } else { + LOG.info( + "Received data statistics from {} operators out of total {} for checkpoint {}. It's less than the expected percentage {}. Thus dropping the incomplete aggregate data statistics {} and starting collecting data statistics from new checkpoint {}", + incompleteAggregateDataStatistics.aggregateSize(), + context.currentParallelism(), + incompleteAggregateDataStatistics.checkpointId(), + EXPECTED_DATA_STATISTICS_RECEIVED_PERCENTAGE, + incompleteAggregateDataStatistics, + checkpointId); + incompleteAggregateDataStatistics = + new AggregateDataStatistics<>(checkpointId, statisticsFactory); + incompleteAggregateDataStatistics.mergeDataStatistic(subtask, event); + } + } else if (incompleteAggregateDataStatistics.checkpointId() > checkpointId) { + LOG.debug( + "Expect data statistics for checkpoint {}, but receive event from older checkpoint {}. Ignore it.", + incompleteAggregateDataStatistics.checkpointId(), + checkpointId); + return; + } else { + incompleteAggregateDataStatistics.mergeDataStatistic(subtask, event); + } + + if (incompleteAggregateDataStatistics.aggregateSize() == context.currentParallelism()) { + completeAggregateDataStatistics = incompleteAggregateDataStatistics; + LOG.info( + "Received data statistics from all {} operators for checkpoint {}. Sending the aggregated data statistics {} to subtasks.", + context.currentParallelism(), + incompleteAggregateDataStatistics.checkpointId(), + completeAggregateDataStatistics); + incompleteAggregateDataStatistics = null; + context.sendDataStatisticsToSubtasks( + checkpointId, completeAggregateDataStatistics.dataStatistics()); + } + } + + @Override + @SuppressWarnings("unchecked") + public void handleEventFromOperator(int subtask, int attemptNumber, OperatorEvent event) { + runInCoordinatorThread( + () -> { + LOG.debug( + "Handling event from subtask {} (#{}) of {}: {}", + subtask, + attemptNumber, + operatorName, + event); + if (event instanceof DataStatisticsEvent) { + handleDataStatisticRequest(subtask, ((DataStatisticsEvent<K>) event)); + } else { + throw new FlinkException("Unrecognized data statistics operator event: " + event); + } + }, + "handling operator event %s from data statistics operator subtask %d (#%d)", + event, + subtask, + attemptNumber); + } + + @Override + public void checkpointCoordinator(long checkpointId, CompletableFuture<byte[]> resultFuture) { + runInCoordinatorThread( + () -> { + LOG.debug( + "Taking a state snapshot on data statistics coordinator {} for checkpoint {}", + operatorName, + checkpointId); + try { + byte[] serializedDataDistributionWeight = Review Comment: discussed offline. either way can work. current way might be more consistent for checking ########## flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsCoordinator.java: ########## @@ -0,0 +1,324 @@ +/* + * 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.iceberg.flink.sink.shuffle; + +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeUnit; +import org.apache.flink.runtime.operators.coordination.OperatorCoordinator; +import org.apache.flink.runtime.operators.coordination.OperatorEvent; +import org.apache.flink.util.ExceptionUtils; +import org.apache.flink.util.FlinkException; +import org.apache.flink.util.InstantiationUtil; +import org.apache.flink.util.Preconditions; +import org.apache.flink.util.function.ThrowingRunnable; +import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting; +import org.jetbrains.annotations.Nullable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * DataStatisticsCoordinator receives {@link DataStatisticsEvent} from {@link + * DataStatisticsOperator} every subtask and then merge them together. Once aggregation for all + * subtasks data statistics completes, DataStatisticsCoordinator will send the aggregation + * result(global data statistics) back to {@link DataStatisticsOperator}. In the end a custom + * partitioner will distribute traffic based on the global data statistics to improve data + * clustering. + */ +class DataStatisticsCoordinator<K> implements OperatorCoordinator { + private static final Logger LOG = LoggerFactory.getLogger(DataStatisticsCoordinator.class); + + private static final double EXPECTED_DATA_STATISTICS_RECEIVED_PERCENTAGE = 0.8; Review Comment: I agree that percentage is more accurate for the purpose here. @yegangy0718 the confusion probably came from the value in the range of 0.0-1.0. When we talk about percentage, people would imagine number like `80` for `80%`. ########## flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsCoordinator.java: ########## @@ -0,0 +1,321 @@ +/* + * 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.iceberg.flink.sink.shuffle; + +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import org.apache.flink.annotation.Internal; +import org.apache.flink.runtime.operators.coordination.OperatorCoordinator; +import org.apache.flink.runtime.operators.coordination.OperatorEvent; +import org.apache.flink.util.ExceptionUtils; +import org.apache.flink.util.InstantiationUtil; +import org.apache.flink.util.Preconditions; +import org.apache.flink.util.function.ThrowingRunnable; +import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting; +import org.jetbrains.annotations.Nullable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * DataStatisticsCoordinator receives {@link DataStatisticsEvent} from {@link + * DataStatisticsOperator} every subtask and then merge them together. Once aggregation for all + * subtasks data statistics completes, DataStatisticsCoordinator will send the aggregated + * result(global data statistics) back to {@link DataStatisticsOperator}. In the end a custom + * partitioner will distribute traffic based on the global data statistics to improve data + * clustering. + */ +@Internal +class DataStatisticsCoordinator<K> implements OperatorCoordinator { + private static final Logger LOG = LoggerFactory.getLogger(DataStatisticsCoordinator.class); + + private static final double EXPECTED_DATA_STATISTICS_RECEIVED_PERCENTAGE = 0.8; + + private final String operatorName; + // A single-thread executor to handle all the actions for coordinator + private final ExecutorService coordinatorExecutor; + private final DataStatisticsCoordinatorContext<K> context; + private final DataStatisticsFactory<K> statisticsFactory; + + private volatile GlobalStatisticsAggregator<K> inProgressAggregator; + private volatile GlobalStatisticsAggregator<K> lastCompletedAggregator; + private volatile boolean started; + + DataStatisticsCoordinator( + String operatorName, + OperatorCoordinator.Context context, + DataStatisticsFactory<K> statisticsFactory) { + this.operatorName = operatorName; + DataStatisticsCoordinatorProvider.CoordinatorExecutorThreadFactory coordinatorThreadFactory = + new DataStatisticsCoordinatorProvider.CoordinatorExecutorThreadFactory( + "DataStatisticsCoordinator-" + operatorName, context.getUserCodeClassloader()); + this.coordinatorExecutor = Executors.newSingleThreadExecutor(coordinatorThreadFactory); + this.context = + new DataStatisticsCoordinatorContext<>( + coordinatorExecutor, coordinatorThreadFactory, context); + this.statisticsFactory = statisticsFactory; + } + + @Override + public void start() throws Exception { + LOG.info("Starting data statistics coordinator for {}.", operatorName); + started = true; + } + + @Override + public void close() throws Exception { + LOG.info("Closing data statistics coordinator for {}.", operatorName); + if (started) { + coordinatorExecutor.shutdown(); + if (!coordinatorExecutor.awaitTermination(5, TimeUnit.SECONDS)) { + LOG.warn("Fail to shut down DataStatisticsCoordinator gracefully. Shutting down now"); + coordinatorExecutor.shutdownNow(); + } + } + } + + private void runInCoordinatorThread( + ThrowingRunnable<Throwable> action, String actionName, Object... actionNameFormatParameters) { + ensureStarted(); + coordinatorExecutor.execute( + () -> { + try { + action.run(); + } catch (Throwable t) { + // if we have a JVM critical error, promote it immediately, there is a good + // chance the logging or job failing will not succeed anymore + ExceptionUtils.rethrowIfFatalErrorOrOOM(t); + + String actionString = String.format(actionName, actionNameFormatParameters); + LOG.error( + "Uncaught exception in the data statistics {} while {}. Triggering job failover.", + operatorName, + actionString, + t); + context.failJob(t); + } + }); + } + + private void ensureStarted() { + if (!this.started) { + throw new IllegalStateException("The coordinator has not started yet."); + } + } + + private void handleDataStatisticRequest(int subtask, DataStatisticsEvent<K> event) { + long checkpointId = event.checkpointId(); + + if (lastCompletedAggregator != null && lastCompletedAggregator.checkpointId() >= checkpointId) { + LOG.debug( + "Data statistics aggregation for checkpoint {} has completed. Ignore the event from subtask {} for checkpoint {}", + lastCompletedAggregator.checkpointId(), + subtask, + checkpointId); + return; + } + + if (inProgressAggregator == null) { + inProgressAggregator = new GlobalStatisticsAggregator<>(checkpointId, statisticsFactory); + } + + if (inProgressAggregator.checkpointId() < checkpointId) { + if ((double) inProgressAggregator.aggregatedSubtasksCount() / context.parallelism() + >= EXPECTED_DATA_STATISTICS_RECEIVED_PERCENTAGE) { + lastCompletedAggregator = inProgressAggregator; + LOG.info( + "Received data statistics from {} operators out of total {} for checkpoint {}. " + + "It's more than the expected percentage {}. Sending the aggregate data" + + " statistics {} to subtasks.", + inProgressAggregator.aggregatedSubtasksCount(), + context.parallelism(), + inProgressAggregator.checkpointId(), + EXPECTED_DATA_STATISTICS_RECEIVED_PERCENTAGE, + lastCompletedAggregator); + inProgressAggregator = new GlobalStatisticsAggregator<>(checkpointId, statisticsFactory); + inProgressAggregator.mergeDataStatistic(subtask, event); + context.sendDataStatisticsToSubtasks( + inProgressAggregator.checkpointId(), lastCompletedAggregator.dataStatistics()); + return; + } else { + LOG.info( + "Received data statistics from {} operators out of total {} for checkpoint {}. " + + "It's less than the expected percentage {}. Dropping the incomplete aggregate " + + "data statistics {} and starting collecting data statistics from new checkpoint {}", + inProgressAggregator.aggregatedSubtasksCount(), + context.parallelism(), + inProgressAggregator.checkpointId(), + EXPECTED_DATA_STATISTICS_RECEIVED_PERCENTAGE, + inProgressAggregator, + checkpointId); + inProgressAggregator = new GlobalStatisticsAggregator<>(checkpointId, statisticsFactory); + } + } else if (inProgressAggregator.checkpointId() > checkpointId) { + LOG.debug( + "Expect data statistics for checkpoint {}, but receive event from older checkpoint {}. Ignore it.", + inProgressAggregator.checkpointId(), + checkpointId); + return; + } + + inProgressAggregator.mergeDataStatistic(subtask, event); + + if (inProgressAggregator.aggregatedSubtasksCount() == context.parallelism()) { + lastCompletedAggregator = inProgressAggregator; + LOG.info( + "Received data statistics from all {} operators for checkpoint {}. Sending the aggregated data statistics {} to subtasks.", + context.parallelism(), + inProgressAggregator.checkpointId(), + lastCompletedAggregator.dataStatistics()); + inProgressAggregator = null; + context.sendDataStatisticsToSubtasks(checkpointId, lastCompletedAggregator.dataStatistics()); + } + } + + @Override + @SuppressWarnings("unchecked") + public void handleEventFromOperator(int subtask, int attemptNumber, OperatorEvent event) { + runInCoordinatorThread( + () -> { + LOG.debug( + "Handling event from subtask {} (#{}) of {}: {}", + subtask, + attemptNumber, + operatorName, + event); + Preconditions.checkArgument(event instanceof DataStatisticsEvent); + handleDataStatisticRequest(subtask, ((DataStatisticsEvent<K>) event)); + }, + "handling operator event %s from data statistics operator subtask %d (#%d)", Review Comment: if this is only logged for error case, the message should be `Failed to process event %s` ########## flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsCoordinatorContext.java: ########## @@ -0,0 +1,172 @@ +/* + * 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.iceberg.flink.sink.shuffle; + +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.Callable; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeUnit; +import org.apache.flink.runtime.operators.coordination.OperatorCoordinator; +import org.apache.flink.shaded.guava30.com.google.common.collect.Iterables; +import org.apache.flink.util.ExceptionUtils; +import org.apache.flink.util.FlinkRuntimeException; +import org.apache.flink.util.Preconditions; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class DataStatisticsCoordinatorContext<K> implements AutoCloseable { + + private static final Logger LOG = LoggerFactory.getLogger(DataStatisticsCoordinatorContext.class); + private final ExecutorService coordinatorExecutor; + private final OperatorCoordinator.Context operatorCoordinatorContext; + private final SubtaskGateways subtaskGateways; + private final DataStatisticsCoordinatorProvider.CoordinatorExecutorThreadFactory + coordinatorThreadFactory; + + DataStatisticsCoordinatorContext( + ExecutorService coordinatorExecutor, + DataStatisticsCoordinatorProvider.CoordinatorExecutorThreadFactory coordinatorThreadFactory, + OperatorCoordinator.Context operatorCoordinatorContext) { + this.coordinatorExecutor = coordinatorExecutor; + this.coordinatorThreadFactory = coordinatorThreadFactory; + this.operatorCoordinatorContext = operatorCoordinatorContext; + this.subtaskGateways = new SubtaskGateways(currentParallelism()); + } + + @Override + public void close() throws Exception { + coordinatorExecutor.shutdown(); + coordinatorExecutor.awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS); + } + + void sendDataStatisticsToSubtasks(long checkpointId, DataStatistics<K> globalDataStatistics) { + callInCoordinatorThread( + () -> { + DataStatisticsEvent<K> dataStatisticsEvent = + new DataStatisticsEvent<>(checkpointId, globalDataStatistics); + int parallelism = currentParallelism(); + for (int i = 0; i < parallelism; ++i) { + subtaskGateways.getOnlyGatewayAndCheckReady(i).sendEvent(dataStatisticsEvent); Review Comment: > The sendEvent is an async call. it returns CompletableFuture. If there is any failure for sending the event, callInCoordinatorThread won't fail. Thx for checking. that leads to a question how to handle async failures here? do we just ignore failures? should we fail the job if there are send failures for a few consecutive cycles? ########## flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/AggregateDataStatistics.java: ########## @@ -0,0 +1,71 @@ +/* + * 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.iceberg.flink.sink.shuffle; + +import java.io.Serializable; +import java.util.Set; +import org.apache.iceberg.relocated.com.google.common.base.MoreObjects; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; +import org.apache.iceberg.relocated.com.google.common.collect.Sets; + +class AggregateDataStatistics<K> implements Serializable { + + private final long checkpointId; + private final DataStatistics<K> dataStatistics; + private final Set<Integer> subtaskSet = Sets.newHashSet(); + + AggregateDataStatistics(long checkpoint, final DataStatisticsFactory<K> statisticsFactory) { + this.checkpointId = checkpoint; + this.dataStatistics = statisticsFactory.createDataStatistics(); + } + + long checkpointId() { + return checkpointId; + } + + DataStatistics<K> dataStatistics() { + return dataStatistics; + } + + void mergeDataStatistic(int subtask, DataStatisticsEvent<K> event) { + Preconditions.checkArgument( Review Comment: discussed offline. caller ensures this won't happen. this is just defensive check ########## flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/sink/shuffle/DataStatisticsCoordinator.java: ########## @@ -0,0 +1,321 @@ +/* + * 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.iceberg.flink.sink.shuffle; + +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import org.apache.flink.annotation.Internal; +import org.apache.flink.runtime.operators.coordination.OperatorCoordinator; +import org.apache.flink.runtime.operators.coordination.OperatorEvent; +import org.apache.flink.util.ExceptionUtils; +import org.apache.flink.util.InstantiationUtil; +import org.apache.flink.util.Preconditions; +import org.apache.flink.util.function.ThrowingRunnable; +import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting; +import org.jetbrains.annotations.Nullable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * DataStatisticsCoordinator receives {@link DataStatisticsEvent} from {@link + * DataStatisticsOperator} every subtask and then merge them together. Once aggregation for all + * subtasks data statistics completes, DataStatisticsCoordinator will send the aggregated + * result(global data statistics) back to {@link DataStatisticsOperator}. In the end a custom + * partitioner will distribute traffic based on the global data statistics to improve data + * clustering. + */ +@Internal +class DataStatisticsCoordinator<K> implements OperatorCoordinator { + private static final Logger LOG = LoggerFactory.getLogger(DataStatisticsCoordinator.class); + + private static final double EXPECTED_DATA_STATISTICS_RECEIVED_PERCENTAGE = 0.8; + + private final String operatorName; + // A single-thread executor to handle all the actions for coordinator + private final ExecutorService coordinatorExecutor; + private final DataStatisticsCoordinatorContext<K> context; + private final DataStatisticsFactory<K> statisticsFactory; + + private volatile GlobalStatisticsAggregator<K> inProgressAggregator; + private volatile GlobalStatisticsAggregator<K> lastCompletedAggregator; + private volatile boolean started; + + DataStatisticsCoordinator( + String operatorName, + OperatorCoordinator.Context context, + DataStatisticsFactory<K> statisticsFactory) { + this.operatorName = operatorName; + DataStatisticsCoordinatorProvider.CoordinatorExecutorThreadFactory coordinatorThreadFactory = + new DataStatisticsCoordinatorProvider.CoordinatorExecutorThreadFactory( + "DataStatisticsCoordinator-" + operatorName, context.getUserCodeClassloader()); + this.coordinatorExecutor = Executors.newSingleThreadExecutor(coordinatorThreadFactory); + this.context = + new DataStatisticsCoordinatorContext<>( + coordinatorExecutor, coordinatorThreadFactory, context); + this.statisticsFactory = statisticsFactory; + } + + @Override + public void start() throws Exception { + LOG.info("Starting data statistics coordinator for {}.", operatorName); + started = true; + } + + @Override + public void close() throws Exception { + LOG.info("Closing data statistics coordinator for {}.", operatorName); + if (started) { + coordinatorExecutor.shutdown(); Review Comment: this is not accurate. the thread pool was created in the constructor. even if the `start` method is not executed, we should still shut down the executor. -- 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]
