ronantakizawa/codereview-bench huggingface dataset:
Repository: apache/flink
File: flink-core/src/main/java/org/apache/flink/api/common/operators/MailOptionsImpl.java (Java)

Code before:
package org.apache.flink.api.common.operators;

import org.apache.flink.annotation.Internal;

/** Options to configure behaviour of executing mailbox mails. */
@Internal
public class MailOptionsImpl implements MailboxExecutor.MailOptions {

    static final MailboxExecutor.MailOptions DEFAULT = new MailOptionsImpl(false, false);
    static final MailboxExecutor.MailOptions DEFERRABLE = new MailOptionsImpl(false, true);
    static final MailboxExecutor.MailOptions HIGH_PRIORITY = new MailOptionsImpl(true, false);

    private final boolean highPriority;
    private final boolean deferrable;

    private MailOptionsImpl(boolean highPriority, boolean deferrable) {
        this.highPriority = highPriority;
        this.deferrable = deferrable;
    }

    public boolean isDeferrable() {
        return deferrable;
    }

    @Override
    public boolean isHighPriority() {
        return highPriority;
    }
}


Review comment (suggestion):
I see here we change the public interface. Do you know why don't we do the same for `isDeferrable` for example? Currently it looks a bit inconsistent

Code after:
package org.apache.flink.api.common.operators;

import org.apache.flink.annotation.Internal;

/** Options to configure behaviour of executing mailbox mails. */
@Internal
public class MailOptionsImpl implements MailboxExecutor.MailOptions {

    static final MailboxExecutor.MailOptions DEFAULT = new MailOptionsImpl(false, false);
    static final MailboxExecutor.MailOptions DEFERRABLE = new MailOptionsImpl(false, true);
    static final MailboxExecutor.MailOptions URGENT = new MailOptionsImpl(true, false);

    private final boolean isUrgent;
    private final boolean deferrable;

    private MailOptionsImpl(boolean isUrgent, boolean deferrable) {
        this.isUrgent = isUrgent;
        this.deferrable = deferrable;
    }

    @Override
    public boolean isDeferrable() {
        return deferrable;
    }

    @Override
    public boolean isUrgent() {
        return isUrgent;
    }
}
Repository: apache/flink
File: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/AbstractCheckpointStats.java (Java)

Code before:
     */
    public abstract CheckpointStatsStatus getStatus();

    /**
     * Returns the number of acknowledged subtasks.
     *
     * @return The number of acknowledged subtasks.
     */
    public abstract int getNumberOfAcknowledgedSubtasks();

    /**
     * Returns the total checkpoint state size over all subtasks.
     *
     * @return Total checkpoint state size over all subtasks.
     */
    public abstract long getStateSize();

    /**
     * Returns the checkpointed size during that checkpoint.
     *
     * @return The checkpointed size during that checkpoint.
     */
    public abstract long getCheckpointedSize();

    /**
     * @return The metadata file size, 0 if unknown.
     */
    public abstract long getMetadataSize();

    /**
     * @return the total number of processed bytes during the checkpoint.
     */
    public abstract long getProcessedData();

    /**
     * @return the total number of persisted bytes during the checkpoint.
     */
    public abstract long getPersistedData();

    /**
     * @return whether the checkpoint is unaligned.
     */
    public abstract boolean isUnalignedCheckpoint();

    /**
     * Returns the latest acknowledged subtask stats or <code>null</code> if none was acknowledged
     * yet.
     *
     * @return Latest acknowledged subtask stats or <code>null</code>
     */
    @Nullable

Review comment (suggestion):
```suggestion
     * @return The metadata file size, -1 if unknown.
```

Code after:
     */
    public abstract CheckpointStatsStatus getStatus();

    /**
     * Returns the number of acknowledged subtasks.
     *
     * @return The number of acknowledged subtasks.
     */
    public abstract int getNumberOfAcknowledgedSubtasks();

    /**
     * Returns the total checkpoint state size over all subtasks.
     *
     * @return Total checkpoint state size over all subtasks.
     */
    public abstract long getStateSize();

    /**
     * Returns the checkpointed size during that checkpoint.
     *
     * @return The checkpointed size during that checkpoint.
     */
    public abstract long getCheckpointedSize();

    /**
     * @return The metadata file size, -1 if unknown.
     */
    public abstract long getMetadataSize();

    /**
     * @return the total number of processed bytes during the checkpoint.
     */
    public abstract long getProcessedData();

    /**
     * @return the total number of persisted bytes during the checkpoint.
     */
    public abstract long getPersistedData();

    /**
     * @return whether the checkpoint is unaligned.
     */
    public abstract boolean isUnalignedCheckpoint();

    /**
     * Returns the latest acknowledged subtask stats or <code>null</code> if none was acknowledged
     * yet.
     *
     * @return Latest acknowledged subtask stats or <code>null</code>
     */
    @Nullable
Repository: apache/flink
File: flink-runtime/src/test/java/org/apache/flink/runtime/blob/NoOpTaskExecutorBlobService.java (Java)

Code before:

package org.apache.flink.runtime.blob;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;

/** No-op {@link TaskExecutorBlobService} implementation for testing purposes. */
public class NoOpTaskExecutorBlobService implements TaskExecutorBlobService {
    public static final NoOpTaskExecutorBlobService INSTANCE = new NoOpTaskExecutorBlobService();

    private NoOpTaskExecutorBlobService() {}

    @Override
    public TransientBlobService getTransientBlobService() {
        return NoOpTransientBlobService.INSTANCE;
    }

    @Override
    public int getPort() {
        return 0;
    }

    @Override
    public InetAddress getAddress() {
        return null;
    }

    @Override
    public JobPermanentBlobService getPermanentBlobService() {
        return NoOpJobPermanentBlobService.INSTANCE;
    }

    @Override
    public void setBlobServerAddress(InetSocketAddress blobServerAddress) {
        // noop
    }

    @Override
    public void close() throws IOException {
        // noop
    }
}


Review comment (bug):
Returning `null` here could cause NullPointerException in future tests.
 Any test code that calls `blobService.getAddress().getHostName()` will throw NPE.
 Since this is a test utility class, it should return a safe dummy value:
Maybe`return InetAddress.getLoopbackAddress();`  ? wdyt.

Code after:

package org.apache.flink.runtime.blob;

import java.io.IOException;
import java.net.InetSocketAddress;

/** No-op {@link TaskExecutorBlobService} implementation for testing purposes. */
public class NoOpTaskExecutorBlobService implements TaskExecutorBlobService {
    public static final NoOpTaskExecutorBlobService INSTANCE = new NoOpTaskExecutorBlobService();

    private NoOpTaskExecutorBlobService() {}

    @Override
    public TransientBlobService getTransientBlobService() {
        return NoOpTransientBlobService.INSTANCE;
    }

    @Override
    public int getPort() {
        return 0;
    }

    @Override
    public JobPermanentBlobService getPermanentBlobService() {
        return NoOpJobPermanentBlobService.INSTANCE;
    }

    @Override
    public void setBlobServerAddress(InetSocketAddress blobServerAddress) {
        // noop
    }

    @Override
    public void close() throws IOException {
        // noop
    }
}
Repository: apache/flink
File: flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/history/retaining/CompositeJobRetainedStrategyTest.java (Java)

Code before:
        @Override
        public long getLen() {
            return 0;
        }

        @Override
        public long getBlockSize() {
            return 0;
        }

        @Override
        public short getReplication() {
            return 0;
        }

        @Override
        public long getModificationTime() {
            return modificationTime;
        }

        @Override
        public long getAccessTime() {
            return 0;
        }

        @Override
        public boolean isDir() {
            return false;
        }

        @Override
        public Path getPath() {
            return null;
        }
    }
}


Review comment (suggestion):
can we test specifying TTL and  no HISTORY_SERVER_RETAINED_JOBS. I think we should honour the TTL in this case, not require the  HISTORY_SERVER_RETAINED_JOBS to have a positive value to have this take effect.

Code after:
    @Test
    void testCompositeBasedJobRetainedStrategy() {

        final long outOfTtlMillis =
                Instant.now().toEpochMilli() - Duration.ofMinutes(2L).toMillis();

        // Test the case where no specific retention policy is configured, i.e., all archived files
        // are retained.
        final Configuration conf = new Configuration();
        JobRetainedStrategy strategy = CompositeJobRetainedStrategy.createFrom(conf);
        assertThat(strategy.shouldRetain(new TestingFileStatus(outOfTtlMillis), 1)).isTrue();
        assertThat(strategy.shouldRetain(new TestingFileStatus(), 10)).isTrue();
        assertThat(strategy.shouldRetain(new TestingFileStatus(outOfTtlMillis), 3)).isTrue();
        assertThat(strategy.shouldRetain(new TestingFileStatus(), 1)).isTrue();

        // Test the case where both retention policies are specified.
        conf.set(HISTORY_SERVER_RETAINED_TTL, Duration.ofMinutes(1));
        conf.set(HISTORY_SERVER_RETAINED_JOBS, 2);
        strategy = CompositeJobRetainedStrategy.createFrom(conf);
        assertThat(strategy.shouldRetain(new TestingFileStatus(outOfTtlMillis), 1)).isFalse();
        assertThat(strategy.shouldRetain(new TestingFileStatus(), 10)).isFalse();
        assertThat(strategy.shouldRetain(new TestingFileStatus(outOfTtlMillis), 3)).isFalse();
        assertThat(strategy.shouldRetain(new TestingFileStatus(), 1)).isTrue();
    }

    private static final class TestingFileStatus implements FileStatus {

        private final long modificationTime;

        TestingFileStatus() {
            this(Instant.now().toEpochMilli());
        }

        TestingFileStatus(long modificationTime) {
            this.modificationTime = modificationTime;
        }

        @Override
        public long getLen() {
            return 0;
        }

        @Override
        public long getBlockSize() {
            return 0;
        }

        @Override
        public short getReplication() {
            return 0;
        }
Repository: apache/flink
File: flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/DeltaJoinTest.scala (Scala)

Code before:
    addTable(
      "tmp_snk",
      Schema
        .newBuilder()
        .column("l0", DataTypes.INT.notNull)
        .column("l1", DataTypes.DOUBLE)
        .primaryKey("l0")
        .build()
    )

    // no joins on this query
    util.verifyRelPlanInsert("insert into tmp_snk select a0, a1 from src1")
  }

  @Test
  def testRetractSink(): Unit = {
    util.tableEnv
      .executeSql(
        "CREATE TABLE retract_snk WITH (\n"
          + "  'sink-changelog-mode-enforced' = 'I,UA,UB,D'"
          + ") LIKE snk (\n"
          + "  OVERWRITING OPTIONS\n"
          + ")")

    util.verifyRelPlanInsert(
      "insert into retract_snk select * from src1 join src2 " +
        "on src1.a1 = src2.b1 " +
        "and src1.a2 = src2.b2")
  }

  @Test
  def testAppendSink(): Unit = {
    util.tableEnv
      .executeSql(
        "CREATE TABLE append_snk WITH (\n"
          + "  'sink-insert-only' = 'true',\n"
          + "  'sink-changelog-mode-enforced' = 'I'\n"
          + ") LIKE snk (\n"
          + "  OVERWRITING OPTIONS\n"
          + ")")

    util.verifyRelPlanInsert(
      "insert into append_snk select * from src1 join src2 " +
        "on src1.a1 = src2.b1 " +
        "and src1.a2 = src2.b2")
  }

  private def addTable(
      tableName: String,
      schema: Schema,
      extraOptions: JMap[String, String] = Collections.emptyMap()): Unit = {

Review comment (suggestion):
Another curious thing is why TABLE_EXEC_SINK_UPSERT_MATERIALIZE should be set to NONE. I have observed that other tests also have the same `upsertMaterialize=[true]` in DeltaJoinTest. xml

Code after:
        "and src1.a2 = src2.b2")

    // one of the joins can be converted into the delta join
    util.verifyExecPlan(stmt)
  }

  @Test
  def testDeltaJoinStrategyWithForce4(): Unit = {
    util.tableConfig.set(
      OptimizerConfigOptions.TABLE_OPTIMIZER_DELTA_JOIN_STRATEGY,
      DeltaJoinStrategy.FORCE)

    addTable(
      "tmp_snk",
      Schema
        .newBuilder()
        .column("l0", DataTypes.INT.notNull)
        .column("l1", DataTypes.DOUBLE)
        .primaryKey("l0")
        .build()
    )

    // no joins on this query
    util.verifyRelPlanInsert("insert into tmp_snk select a0, a1 from src1")
  }

  @Test
  def testRetractSink(): Unit = {
    util.tableEnv
      .executeSql(
        "CREATE TABLE retract_snk WITH (\n"
          + "  'sink-changelog-mode-enforced' = 'I,UA,UB,D'"
          + ") LIKE snk (\n"
          + "  OVERWRITING OPTIONS\n"
          + ")")

    util.verifyRelPlanInsert(
      "insert into retract_snk select * from src1 join src2 " +
        "on src1.a1 = src2.b1 " +
        "and src1.a2 = src2.b2")
  }

  @Test
  def testAppendSink(): Unit = {
    util.tableEnv
      .executeSql(
        "CREATE TABLE append_snk WITH (\n"
          + "  'sink-insert-only' = 'true',\n"
          + "  'sink-changelog-mode-enforced' = 'I'\n"
          + ") LIKE snk (\n"
          + "  OVERWRITING OPTIONS\n"
Repository: apache/flink
File: flink-runtime/src/main/java/org/apache/flink/runtime/state/AsyncKeyedStateBackend.java (Java)

Code before:
     * @return true as default in case of AsyncKeyedStateBackend
     */
    default boolean requiresLegacySynchronousTimerSnapshots(SnapshotType checkpointType) {
        return true;
    }

    /**
     * Whether it's safe to reuse key-values from the state-backend, e.g for the purpose of
     * optimization.
     *
     * <p>NOTE: this method should not be used to check for {@link InternalPriorityQueue}, as the
     * priority queue could be stored on different locations, e.g ForSt state-backend could store
     * that on JVM heap if configuring HEAP as the time-service factory.
     *
     * @return returns ture if safe to reuse the key-values from the state-backend.
     */
    default boolean isSafeToReuseKVState() {
        return false;
    }

    @Override
    void dispose();

    /**
     * @return fixed lower-case string identifying the type of the underlying state backend, e.g.
     *     rocksdb, hashmap, or unknown.
     */
    @Experimental
    default String getBackendTypeIdentifier() {
        return "unknown";
    }
}


Review comment (suggestion):
Is forst missed?

```suggestion
     *     rocksdb, hashmap, forst or unknown.
```

Code after:
     * @return true as default in case of AsyncKeyedStateBackend
     */
    default boolean requiresLegacySynchronousTimerSnapshots(SnapshotType checkpointType) {
        return true;
    }

    /**
     * Whether it's safe to reuse key-values from the state-backend, e.g for the purpose of
     * optimization.
     *
     * <p>NOTE: this method should not be used to check for {@link InternalPriorityQueue}, as the
     * priority queue could be stored on different locations, e.g ForSt state-backend could store
     * that on JVM heap if configuring HEAP as the time-service factory.
     *
     * @return returns ture if safe to reuse the key-values from the state-backend.
     */
    default boolean isSafeToReuseKVState() {
        return false;
    }

    @Override
    void dispose();

    /**
     * @return fixed lower-case string identifying the type of the underlying state backend, e.g.
     *     rocksdb, hashmap, or unknown.
     */
    @Experimental
    String getBackendTypeIdentifier();
}
Repository: apache/flink
File: flink-core/src/main/java/org/apache/flink/api/common/state/KeyedStateStore.java (Java)

Code before:
     * Gets a handle to the system's key/value aggregating state. This state is only accessible if
     * the function is executed on a KeyedStream.
     *
     * @param stateProperties The descriptor defining the properties of the stats.
     * @param <IN> The type of the values that are added to the state.
     * @param <ACC> The type of the accumulator (intermediate aggregation state).
     * @param <OUT> The type of the values that are returned from the state.
     * @return The partitioned state object.
     * @throws UnsupportedOperationException Thrown, if no partitioned state is available for the
     *     function (function is not part of a KeyedStream).
     */
    @Experimental
    <IN, ACC, OUT>
            org.apache.flink.api.common.state.v2.AggregatingState<IN, OUT> getAggregatingState(
                    @Nonnull
                            org.apache.flink.api.common.state.v2.AggregatingStateDescriptor<
                                            IN, ACC, OUT>
                                    stateProperties);

    /**
     * @return fixed lower-case string identifying the type of the underlying state backend, e.g.
     *     rocksdb, hashmap, forst, batch, or unknown.
     */
    @Experimental
    default String getBackendTypeIdentifier() {
        return "unknown";
    }
}


Review comment (suggestion):
nit: How about removing the default implementation? This would prevent subclasses from forgetting to override `getBackendTypeIdentifier`.

Code after:
     * Gets a handle to the system's key/value aggregating state. This state is only accessible if
     * the function is executed on a KeyedStream.
     *
     * @param stateProperties The descriptor defining the properties of the stats.
     * @param <IN> The type of the values that are added to the state.
     * @param <ACC> The type of the accumulator (intermediate aggregation state).
     * @param <OUT> The type of the values that are returned from the state.
     * @return The partitioned state object.
     * @throws UnsupportedOperationException Thrown, if no partitioned state is available for the
     *     function (function is not part of a KeyedStream).
     */
    @Experimental
    <IN, ACC, OUT>
            org.apache.flink.api.common.state.v2.AggregatingState<IN, OUT> getAggregatingState(
                    @Nonnull
                            org.apache.flink.api.common.state.v2.AggregatingStateDescriptor<
                                            IN, ACC, OUT>
                                    stateProperties);

    /**
     * @return fixed lower-case string identifying the type of the underlying state backend, e.g.
     *     rocksdb, hashmap, forst, batch.
     */
    @Experimental
    String getBackendTypeIdentifier();
}
Repository: apache/flink
File: flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/util/stats/StatsSummaryDto.java (Java)

Code before:

    public double getP95() {
        return p95;
    }

    public double getP99() {
        return p99;
    }

    public double getP999() {
        return p999;
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) {
            return true;
        }
        if (o == null || getClass() != o.getClass()) {
            return false;
        }
        StatsSummaryDto that = (StatsSummaryDto) o;
        return minimum == that.minimum
                && maximum == that.maximum
                && average == that.average
                && p50 == that.p50
                && p90 == that.p90
                && p95 == that.p95
                && p99 == that.p99
                && p999 == that.p999;
    }

    @Override
    public int hashCode() {
        return Objects.hash(minimum, maximum, average, p50, p90, p95, p99, p999);
    }
}


Review comment (suggestion):
I am thinking that NaN should never be in the stats, would a better root cause fix be to police the code, so that NaN is never written. I assume NaN does not mean anything in this context. It would be worth adding this code as it seems reasonable to self protect the code. So i will approve.

Code after:

    public double getP95() {
        return p95;
    }

    public double getP99() {
        return p99;
    }

    public double getP999() {
        return p999;
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) {
            return true;
        }
        if (o == null || getClass() != o.getClass()) {
            return false;
        }
        StatsSummaryDto that = (StatsSummaryDto) o;
        return minimum == that.minimum
                && maximum == that.maximum
                && average == that.average
                && Double.compare(p50, that.p50) == 0
                && Double.compare(p90, that.p90) == 0
                && Double.compare(p95, that.p95) == 0
                && Double.compare(p99, that.p99) == 0
                && Double.compare(p999, that.p999) == 0;
    }

    @Override
    public int hashCode() {
        return Objects.hash(minimum, maximum, average, p50, p90, p95, p99, p999);
    }
}
Repository: apache/flink
File: flink-runtime/src/main/java/org/apache/flink/streaming/runtime/tasks/SystemProcessingTimeService.java (Java)

Code before:
                long timestamp,
                long period,
                boolean fixedDelay) {
            this.serviceStatus = serviceStatus;
            this.exceptionHandler = exceptionHandler;
            this.callback = callback;
            this.nextTimestamp = timestamp;
            this.period = period;
            this.fixedDelay = fixedDelay;
        }

        @Override
        public void run() {
            if (serviceStatus.get() != STATUS_ALIVE) {
                return;
            }
            try {
                callback.onProcessingTime(nextTimestamp);
            } catch (Exception ex) {
                exceptionHandler.handleException(ex);
            }
            if (fixedDelay) {
                nextTimestamp = System.currentTimeMillis() + period;
            } else {
                nextTimestamp += period;
            }
        }
    }
}


Review comment (suggestion):
```suggestion
            nextTimestamp =
                    fixedDelay ? System.currentTimeMillis() + period : nextTimestamp + period;
```

Code after:
                long timestamp,
                long period,
                boolean fixedDelay) {
            this.serviceStatus = serviceStatus;
            this.exceptionHandler = exceptionHandler;
            this.callback = callback;
            this.nextTimestamp = timestamp;
            this.period = period;
            this.fixedDelay = fixedDelay;
        }

        @Override
        public void run() {
            if (serviceStatus.get() != STATUS_ALIVE) {
                return;
            }
            try {
                callback.onProcessingTime(nextTimestamp);
            } catch (Exception ex) {
                exceptionHandler.handleException(ex);
            }
            nextTimestamp =
                    fixedDelay ? System.currentTimeMillis() + period : nextTimestamp + period;
        }
    }
}