ronantakizawa/codereview-bench huggingface dataset:
Repository: apache/arrow
File: ci/scripts/python_wheel_validate_contents.py (Python)

Code before:
# 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.

import argparse
from pathlib import Path
import re
import zipfile


def validate_wheel(path):
    p = Path(path)
    wheels = list(p.glob('*.whl'))
    error_msg = f"{len(wheels)} wheels found but only 1 expected ({wheels})"
    assert len(wheels) == 1, error_msg
    f = zipfile.ZipFile(wheels[0])
    outliers = [
        info.filename for info in f.filelist if not re.match(
            r'(pyarrow/|pyarrow-[-.\w\d]+\.dist-info/|pyarrow\.libs/)', info.filename
        )
    ]
    assert not outliers, f"Unexpected contents in wheel: {sorted(outliers)}"
    for filename in ('LICENSE.txt', 'NOTICE.txt'):
        assert any(info.filename.endswith(filename) for info in f.filelist), \
            f"{filename} is missing from the wheel."
    print(f"The wheel: {wheels[0]} seems valid.")


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--path", type=str, required=True,
                        help="Directory where wheel is located")
    args = parser.parse_args()
    validate_wheel(args.path)


if __name__ == '__main__':
    main()


Review comment (suggestion):
How about being a bit stricter
```suggestion
        assert any(info.filename.split("/")[-1] == filename
                   for info in f.filelist), \
```

Code after:
# 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.

import argparse
from pathlib import Path
import re
import zipfile


def validate_wheel(path):
    p = Path(path)
    wheels = list(p.glob('*.whl'))
    error_msg = f"{len(wheels)} wheels found but only 1 expected ({wheels})"
    assert len(wheels) == 1, error_msg
    f = zipfile.ZipFile(wheels[0])
    outliers = [
        info.filename for info in f.filelist if not re.match(
            r'(pyarrow/|pyarrow-[-.\w\d]+\.dist-info/|pyarrow\.libs/)', info.filename
        )
    ]
    assert not outliers, f"Unexpected contents in wheel: {sorted(outliers)}"
    for filename in ('LICENSE.txt', 'NOTICE.txt'):
        assert any(info.filename.split("/")[-1] == filename
                   for info in f.filelist), \
            f"{filename} is missing from the wheel."
    print(f"The wheel: {wheels[0]} seems valid.")


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--path", type=str, required=True,
                        help="Directory where wheel is located")
    args = parser.parse_args()
    validate_wheel(args.path)


if __name__ == '__main__':
    main()
Repository: apache/arrow
File: r/tests/testthat/test-dataset-write.R (R)

Code before:
  expect_equal(lines[2], "\"1\"\t\"1\"\t\"true\"\t\"a\"")
  dst_dir <- make_temp_dir()
  write_tsv_dataset(df, dst_dir, na = "NOVALUE")
  ds <- open_dataset(dst_dir, format = "tsv") |> collect()

  expect_equal(
    ds$lgl,
    c("true", "false", "NOVALUE", "true", "false", "true", "false", "NOVALUE", "true", "false")
  )
})

test_that("Row order is preserved when writing large parquet dataset", {
  skip_if_not_available("parquet")
  # Make a data frame with a sufficiently large number of rows.
  df <- data.frame(x = 1:1.1e6)
  dst_dir1 <- make_temp_dir()
  write_dataset(df, dst_dir1)

  dst_dir2 <- make_temp_dir()
  write_dataset(df, dst_dir2, preserve_order = TRUE)

  ds1 <- open_dataset(dst_dir1) |> collect()
  ds2 <- open_dataset(dst_dir2) |> collect()

  expect_true(any(ds1$x != df$x))
  expect_true(all(ds2$x == df$x))
})


Review comment (suggestion):
```suggestion
  unordered_dir <- make_temp_dir()
  write_dataset(df, unordered_dir)

  ordered_dir <- make_temp_dir()
  write_dataset(df, ordered_dir, preserve_order = TRUE)

  unordered_ds <- open_dataset(unordered_dir) |> collect()
  ordered_ds2 <- open_dataset(ordered_dir) |> collect()

  expect_true(any(unordered_ds$x != df$x))
  expect_true(all(ordered_ds$x == df$x))
```

Making the names slightly more understandable / readable. 

also what about:

```
# unordered is setequal, but not ordered
expect_setequal(unordered_ds$x, df$x)
expect_false(all(unordered_ds$x == df$x))

# but ordered is exactly equal
expect_equal(ordered_ds$x, df$x)
```

Code after:
  expect_equal(lines[2], "\"1\"\t\"1\"\t\"true\"\t\"a\"")
  dst_dir <- make_temp_dir()
  write_tsv_dataset(df, dst_dir, na = "NOVALUE")
  ds <- open_dataset(dst_dir, format = "tsv") |> collect()

  expect_equal(
    ds$lgl,
    c("true", "false", "NOVALUE", "true", "false", "true", "false", "NOVALUE", "true", "false")
  )
})

test_that("Row order is preserved when writing large parquet dataset", {
  skip_if_not_available("parquet")
  # Make a data frame with a sufficiently large number of rows.
  df <- data.frame(x = 1:1.1e6)

  unordered_dir <- make_temp_dir()
  write_dataset(df, unordered_dir)

  ordered_dir <- make_temp_dir()
  write_dataset(df, ordered_dir, preserve_order = TRUE)

  unordered_ds <- open_dataset(unordered_dir) |> collect()
  ordered_ds <- open_dataset(ordered_dir) |> collect()

  # Unordered is set equal, but not necessarily equal.
  expect_setequal(unordered_ds$x, df$x)
  # expect_false(all(unordered_ds$x == df$x)) can fail on certain
  # platforms, so is not tested.

  # But ordered is exactly equal.
  expect_equal(ordered_ds$x, df$x)
})
Repository: apache/dubbo
File: dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/bootstrap/DubboBootstrap.java (Java)

Code before:
                }
            }
        });
    }

    private void shutdown() {
        if (!executorService.isShutdown()) {
            // Shutdown executorService
            executorService.shutdown();
        }
    }

    private void executeMutually(Runnable runnable) {
        try {
            lock.lock();
            runnable.run();
        } finally {
            lock.unlock();
        }
    }

    public ApplicationConfig getApplication() {
        return configManager.getApplicationOrElseThrow();
    }

    public void setTakeoverState(BootstrapTakeoverMode takeoverState) {
        this.takeoverState = takeoverState;
    }

    public BootstrapTakeoverMode getTakeoverState() {
        return takeoverState;
    }
}


Review comment (suggestion):
Please rename method `setTakeoverState` to `setTakeoverMode`



Code after:
                }
            }
        });
    }

    private void shutdown() {
        if (!executorService.isShutdown()) {
            // Shutdown executorService
            executorService.shutdown();
        }
    }

    private void executeMutually(Runnable runnable) {
        try {
            lock.lock();
            runnable.run();
        } finally {
            lock.unlock();
        }
    }

    public ApplicationConfig getApplication() {
        return configManager.getApplicationOrElseThrow();
    }

    public void setTakeoverMode(BootstrapTakeoverMode takeoverMode) {
        this.takeoverMode = takeoverMode;
    }

    public BootstrapTakeoverMode getTakeoverMode() {
        return takeoverMode;
    }
}
Repository: apache/dubbo
File: dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/deploy/DefaultApplicationDeployer.java (Java)

Code before:
        return serviceInstance;
    }

    @Override
    public void stop() {
        destroy();
    }

    @Override
    public synchronized void destroy() {
        if (isStopping() || isStopped()) {
            return;
        }
        try {
            onStopping();
            unRegisterShutdownHook();
            unregisterServiceInstance();
            unexportMetadataService();

            executeShutdownCallbacks();

            applicationModel.destroy();
            if (asyncMetadataFuture != null) {
                asyncMetadataFuture.cancel(true);
            }

            destroyProtocols();

            destroyRegistries();
            destroyServiceDiscoveries();
            destroyMetadataReports();

            destroyServiceDiscoveries();
            destroyExecutorRepository();
            destroyDynamicConfigurations();

            onStopped();
        } catch (Throwable ex) {
            logger.error(getIdentifier() + " an error occurred while stopping application: " + ex.getMessage(), ex);
            setFailed(ex);
        }
    }

    private void executeShutdownCallbacks() {
        ShutdownHookCallbacks shutdownHookCallbacks = applicationModel.getBeanFactory().getBean(ShutdownHookCallbacks.class);
        shutdownHookCallbacks.callback();
    }

    @Override
    public void checkStarting() {
        if (isStarting()) {

Review comment (suggestion):
Metadata refresh future should be canceled after unregister service instance, which should be called before `applicationModel.destroy()`.

Code after:
        return serviceInstance;
    }

    @Override
    public void stop() {
        destroy();
    }

    @Override
    public synchronized void destroy() {
        if (isStopping() || isStopped()) {
            return;
        }
        try {
            onStopping();
            unRegisterShutdownHook();
            unregisterServiceInstance();
            unexportMetadataService();
            if (asyncMetadataFuture != null) {
                asyncMetadataFuture.cancel(true);
            }

            executeShutdownCallbacks();

            applicationModel.destroy();

            destroyProtocols();

            destroyRegistries();
            destroyServiceDiscoveries();
            destroyMetadataReports();

            destroyServiceDiscoveries();
            destroyExecutorRepository();
            destroyDynamicConfigurations();

            onStopped();
        } catch (Throwable ex) {
            logger.error(getIdentifier() + " an error occurred while stopping application: " + ex.getMessage(), ex);
            setFailed(ex);
        }
    }

    private void executeShutdownCallbacks() {
        ShutdownHookCallbacks shutdownHookCallbacks = applicationModel.getBeanFactory().getBean(ShutdownHookCallbacks.class);
        shutdownHookCallbacks.callback();
    }

    @Override
    public void checkStarting() {
        if (isStarting()) {
Repository: apache/dubbo
File: dubbo-plugin/dubbo-qos/src/main/java/org/apache/dubbo/qos/command/impl/DefaultMetricsReporterCmd.java (Java)

Code before:
            return "Not exist application: " + appName;
        }
    }

    private String allApplication(List<ApplicationModel> models) {
        Map<String, String> appResultMap = new HashMap<>();
        for (ApplicationModel model : models) {
            appResultMap.put(model.getApplicationName(), getResponseByApplication(model, null));
        }
        return JsonUtils.toJson(appResultMap);
    }

    private String getResponseByApplication(ApplicationModel applicationModel, String metricsName) {
        String response = "MetricsReporter not init";
        MetricsReporter metricsReporter = applicationModel.getBeanFactory().getBean(MetricsReporter.class);
        if (metricsReporter != null) {
            metricsReporter.refreshData();
            response = metricsReporter.getResponseWithName(metricsName);
        }
        return response;
    }


    private static long getLineNumber(String content) {
        LineNumberReader lnr = new LineNumberReader(new CharArrayReader(content.toCharArray()));
        try {
            lnr.skip(Long.MAX_VALUE);
            lnr.close();
        } catch (IOException ignore) {
        }
        return lnr.getLineNumber();
    }
}


Review comment (suggestion):
please fix NullPointException
<img width="1178" alt="image" src="https://github.com/apache/dubbo/assets/81170548/f8845a5c-8444-478f-9020-6e7728c2b3b3">


Code after:
        } else {
            return "Not exist application: " + appName;
        }
    }

    private String allApplication(List<ApplicationModel> models) {
        Map<String, String> appResultMap = new HashMap<>();
        for (ApplicationModel model : models) {
            appResultMap.put(model.getApplicationName(), getResponseByApplication(model, null));
        }
        return JsonUtils.toJson(appResultMap);
    }

    private String getResponseByApplication(ApplicationModel applicationModel, String metricsName) {
        String response = "DefaultMetricsReporter not init";
        MetricsReporter metricsReporter = applicationModel.getBeanFactory().getBean(DefaultMetricsReporter.class);
        if (metricsReporter != null) {
            metricsReporter.refreshData();
            response = metricsReporter.getResponseWithName(metricsName);
        }
        return response;
    }


    private static long getLineNumber(String content) {
        LineNumberReader lnr = new LineNumberReader(new CharArrayReader(content.toCharArray()));
        try {
            lnr.skip(Long.MAX_VALUE);
            lnr.close();
        } catch (IOException ignore) {
        }
        return lnr.getLineNumber();
    }
}
Repository: apache/dubbo
File: dubbo-common/src/main/java/org/apache/dubbo/common/utils/LockUtils.java (Java)

Code before:
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.locks.Lock;

public class LockUtils {
    private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(LockUtils.class);

    public static final int DEFAULT_TIMEOUT = 60_000;

    public static void safeLock(Lock lock, int timeout, Runnable runnable) {
        try {
            if (!lock.tryLock(timeout, TimeUnit.MILLISECONDS)) {
                logger.error(
                        LoggerCodeConstants.INTERNAL_ERROR,
                        "",
                        "",
                        "Try to lock failed, timeout: " + timeout,
                        new TimeoutException());
            }
            runnable.run();
        } catch (InterruptedException e) {
            logger.warn(LoggerCodeConstants.INTERNAL_ERROR, "", "", "Try to lock failed", e);
            Thread.currentThread().interrupt();
        } finally {
            try {
                lock.unlock();
            } catch (Exception e) {
                // ignore
            }
        }
    }
}


Review comment (bug):
This may seem a bit strange, because generally the tryLock method is used like this:
```java
 Lock lock = ...;
  if (lock.tryLock()) {
      try {
          // do something with lock
          runnable.run();
      } finally {
          lock.unlock();
      }
  } 
```

Wouldn't this really lead to some race conditions?

Code after:
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.locks.Lock;

public class LockUtils {
    private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(LockUtils.class);

    public static final int DEFAULT_TIMEOUT = 60_000;

    public static void safeLock(Lock lock, int timeout, Runnable runnable) {
        try {
            boolean interrupted = false;
            try {
                if (!lock.tryLock(timeout, TimeUnit.MILLISECONDS)) {
                    logger.error(
                            LoggerCodeConstants.INTERNAL_ERROR,
                            "",
                            "",
                            "Try to lock failed, timeout: " + timeout,
                            new TimeoutException());
                }
            } catch (InterruptedException e) {
                logger.warn(LoggerCodeConstants.INTERNAL_ERROR, "", "", "Try to lock failed", e);
                interrupted = true;
            }
            runnable.run();
            if (interrupted) {
                Thread.currentThread().interrupt();
            }
        } finally {
            try {
                lock.unlock();
            } catch (Exception e) {
                // ignore
            }
        }
    }
}
Repository: apache/dubbo
File: dubbo-common/src/main/java/org/apache/dubbo/common/utils/UrlUtils.java (Java)

Code before:
                // see https://howdoesinternetwork.com/2013/ipv6-zone-id
                // ignore
            } else {
                port = Integer.parseInt(url.substring(i + 1));
                url = url.substring(0, i);
            }
        }
        if (url.length() > 0) {
            host = url;
        }

        return new ServiceConfigURL(protocol, username, password, host, port, path, parameters);
    }

    public static boolean isConsumer(URL url) {
        return url.getProtocol().equalsIgnoreCase(CONSUMER) || url.getPort() == 0;
    }

    @SuppressWarnings("unchecked")
    public static <T> T computeServiceAttribute(URL url, String key, Function<URL, T> fn) {
        Optional<Map<String, Object>> attributeMap = Optional.ofNullable(url.getServiceModel())
                .map(ServiceModel::getServiceMetadata)
                .map(ServiceMetadata::getAttributeMap);
        return attributeMap
                .map(stringObjectMap -> (T) stringObjectMap.computeIfAbsent(key, k -> fn.apply(url)))
                .orElse(null);
    }
}


Review comment (suggestion):
```suggestion
        return Optional.ofNullable(url.getServiceModel())
                .map(ServiceModel::getServiceMetadata)
                .map(ServiceMetadata::getAttributeMap)
                .map(stringObjectMap -> (T) stringObjectMap.computeIfAbsent(key, k -> fn.apply(url)))
                .orElse(null);
```

Code after:
                // see https://howdoesinternetwork.com/2013/ipv6-zone-id
                // ignore
            } else {
                port = Integer.parseInt(url.substring(i + 1));
                url = url.substring(0, i);
            }
        }
        if (url.length() > 0) {
            host = url;
        }

        return new ServiceConfigURL(protocol, username, password, host, port, path, parameters);
    }

    public static boolean isConsumer(URL url) {
        return url.getProtocol().equalsIgnoreCase(CONSUMER) || url.getPort() == 0;
    }

    @SuppressWarnings("unchecked")
    public static <T> T computeServiceAttribute(URL url, String key, Function<URL, T> fn) {
        return Optional.ofNullable(url.getServiceModel())
                .map(ServiceModel::getServiceMetadata)
                .map(ServiceMetadata::getAttributeMap)
                .map(stringObjectMap -> (T) stringObjectMap.computeIfAbsent(key, k -> fn.apply(url)))
                .orElse(null);
    }
}
Repository: apache/dubbo
File: dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/http1/Http1SseServerChannelObserver.java (Java)

Code before:
 * limitations under the License.
 */
package org.apache.dubbo.rpc.protocol.tri.h12.http1;

import org.apache.dubbo.remoting.http12.HttpChannel;
import org.apache.dubbo.remoting.http12.HttpConstants;
import org.apache.dubbo.remoting.http12.HttpHeaderNames;
import org.apache.dubbo.remoting.http12.HttpMetadata;
import org.apache.dubbo.remoting.http12.h1.Http1ServerChannelObserver;
import org.apache.dubbo.remoting.http12.message.HttpMessageEncoder;
import org.apache.dubbo.remoting.http12.message.ServerSentEventEncoder;

public class Http1SseServerChannelObserver extends Http1ServerChannelObserver {

    public Http1SseServerChannelObserver(HttpChannel httpChannel) {
        super(httpChannel);
    }

    @Override
    public void setResponseEncoder(HttpMessageEncoder responseEncoder) {
        super.setResponseEncoder(new ServerSentEventEncoder(responseEncoder));
    }

    @Override
    protected void doOnCompleted(Throwable throwable) {
        sendMetadata(encodeHttpMetadata(true));
        super.doOnCompleted(throwable);
    }

    @Override
    protected HttpMetadata encodeHttpMetadata(boolean endStream) {
        return super.encodeHttpMetadata(endStream)
                .header(HttpHeaderNames.TRANSFER_ENCODING.getKey(), HttpConstants.CHUNKED)
                .header(HttpHeaderNames.CACHE_CONTROL.getKey(), HttpConstants.NO_CACHE);
    }
}


Review comment (suggestion):
Your previous approach was correct — a check is needed here to avoid an unnesssory encoding of Metadata

Code after:
 * limitations under the License.
 */
package org.apache.dubbo.rpc.protocol.tri.h12.http1;

import org.apache.dubbo.remoting.http12.HttpChannel;
import org.apache.dubbo.remoting.http12.HttpConstants;
import org.apache.dubbo.remoting.http12.HttpHeaderNames;
import org.apache.dubbo.remoting.http12.HttpMetadata;
import org.apache.dubbo.remoting.http12.h1.Http1ServerChannelObserver;
import org.apache.dubbo.remoting.http12.message.HttpMessageEncoder;
import org.apache.dubbo.remoting.http12.message.ServerSentEventEncoder;

public class Http1SseServerChannelObserver extends Http1ServerChannelObserver {

    public Http1SseServerChannelObserver(HttpChannel httpChannel) {
        super(httpChannel);
    }

    @Override
    public void setResponseEncoder(HttpMessageEncoder responseEncoder) {
        super.setResponseEncoder(new ServerSentEventEncoder(responseEncoder));
    }

    @Override
    protected void doOnCompleted(Throwable throwable) {
        if (!isHeaderSent()) {
            sendMetadata(encodeHttpMetadata(true));
        }
        super.doOnCompleted(throwable);
    }

    @Override
    protected HttpMetadata encodeHttpMetadata(boolean endStream) {
        return super.encodeHttpMetadata(endStream)
                .header(HttpHeaderNames.TRANSFER_ENCODING.getKey(), HttpConstants.CHUNKED)
                .header(HttpHeaderNames.CACHE_CONTROL.getKey(), HttpConstants.NO_CACHE);
    }
}
Repository: apache/flink
File: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/lineage/TypeDatasetFacet.java (Java)

Code before:
/*
 * 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.flink.streaming.api.lineage;

import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.api.common.typeinfo.TypeInformation;

/** Facet definition to contain type information of source and sink. */
@PublicEvolving
public interface TypeDatasetFacet extends LineageDatasetFacet {

    TypeInformation getTypeInformation();
}


Review comment (suggestion):
NIT: can we add @nonnull and @PublicEvolving to this method. I see examples in the Flink codebase where @PublicEvolving is on the methods as well as the class.  

If it can be null it should be annotated with @nullable

Code after:
 * 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.flink.streaming.api.lineage;

import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.api.common.typeinfo.TypeInformation;

import javax.annotation.Nonnull;

/** Facet definition to contain type information of source and sink. */
@PublicEvolving
public interface TypeDatasetFacet extends LineageDatasetFacet {

    @Nonnull
    TypeInformation getTypeInformation();
}