ronantakizawa/codereview-bench huggingface dataset:
Repository: apache/flink
File: flink-table/flink-table-common/src/test/java/org/apache/flink/types/ColumnListTest.java (Java)

Code before:

    @Test
    void testNullElementInDataTypesThrowsException() {
        final List<String> names = Arrays.asList("a", "b", "c");
        final List<DataType> dataTypes = Arrays.asList(DataTypes.INT(), null, DataTypes.BOOLEAN());

        assertThatThrownBy(() -> ColumnList.of(names, dataTypes))
                .isInstanceOf(IllegalArgumentException.class)
                .hasMessageContaining("Data types must not contain null elements");
    }

    @Test
    void testImmutabilityOfNames() {
        final List<String> names = new ArrayList<>(Arrays.asList("a", "b"));
        final ColumnList columnList = ColumnList.of(names);

        names.add("c");

        assertThat(columnList.getNames()).containsExactly("a", "b");
    }

    @Test
    void testImmutabilityOfDataTypes() {
        final List<String> names = new ArrayList<>(Arrays.asList("a", "b"));
        final List<DataType> dataTypes =
                new ArrayList<>(Arrays.asList(DataTypes.INT(), DataTypes.STRING()));
        final ColumnList columnList = ColumnList.of(names, dataTypes);

        dataTypes.add(DataTypes.BOOLEAN());

        assertThat(columnList.getDataTypes()).hasSize(2);
    }
}


Review comment (suggestion):
```suggestion
                List.of(DataTypes.INT(), DataTypes.STRING()));
```

Code after:
    void testNullElementInDataTypesThrowsException() {
        final List<String> names = List.of("a", "b", "c");
        final List<DataType> dataTypes = Arrays.asList(DataTypes.INT(), null, DataTypes.BOOLEAN());

        assertThatThrownBy(() -> ColumnList.of(names, dataTypes))
                .isInstanceOf(IllegalArgumentException.class)
                .hasMessageContaining("Data types must not contain null elements");
    }

    @Test
    void testImmutabilityOfNames() {
        final List<String> names = new ArrayList<>(List.of("a", "b"));
        final ColumnList columnList = ColumnList.of(names);

        names.add("c");

        assertThat(columnList.getNames()).containsExactly("a", "b");
    }

    @Test
    void testImmutabilityOfDataTypes() {
        final List<String> names = new ArrayList<>(List.of("a", "b"));
        final List<DataType> dataTypes =
                new ArrayList<>(List.of(DataTypes.INT(), DataTypes.STRING()));
        final ColumnList columnList = ColumnList.of(names, dataTypes);

        dataTypes.add(DataTypes.BOOLEAN());

        assertThat(columnList.getDataTypes()).hasSize(2);
    }
}
Repository: apache/kafka
File: streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamFlatTransformValuesTest.java (Java)

Code before:

        processor.process(new Record<>(inputKey, inputValue, 0L));

        verify(context, never()).forward(ArgumentMatchers.<Record<Integer, String>>any());
    }

    @Test
    public void shouldCloseFlatTransformValuesProcessor() {
        processor.close();

        verify(valueTransformer).close();
    }

    @Test
    public void shouldGetFlatTransformValuesProcessor() {
        @SuppressWarnings("unchecked")
        final ValueTransformerWithKeySupplier<Integer, Integer, Iterable<String>> valueTransformerSupplier =
            mock(ValueTransformerWithKeySupplier.class);
        final KStreamFlatTransformValues<Integer, Integer, String> processorSupplier =
            new KStreamFlatTransformValues<>(valueTransformerSupplier);

        when(valueTransformerSupplier.get()).thenReturn(valueTransformer);

        final Processor<Integer, Integer, Integer, String> processor = processorSupplier.get();

        verify(valueTransformerSupplier).get();
        assertTrue(processor instanceof KStreamFlatTransformValuesProcessor);
    }
}


Review comment (suggestion):
[optional]

I see that sometimes we explicitly verify stubbed actions and I'm wondering if there is a rule/guideline on when to do that (`stub` + `verify`) 🤔 

I'm asking because according to the [Mockito docs](https://javadoc.io/doc/org.mockito/mockito-core/latest/org/mockito/Mockito.html#stubbing), if we stub something while using `Strict Stubbing mode`, we get verify part out-of-the-box.

<img width="597" alt="image" src="https://user-images.githubusercontent.com/9079844/184425984-5a23e722-83d0-4104-9428-dde40ba4fd67.png">


Code after:
    public void shouldEmitNoRecordIfTransformReturnsNull() {
        processor.init(context);

        when(valueTransformer.transform(inputKey, inputValue)).thenReturn(null);

        processor.process(new Record<>(inputKey, inputValue, 0L));

        inOrder.verify(context, never()).forward(ArgumentMatchers.<Record<Integer, String>>any());
    }

    @Test
    public void shouldCloseFlatTransformValuesProcessor() {
        processor.close();

        verify(valueTransformer).close();
    }

    @Test
    public void shouldGetFlatTransformValuesProcessor() {
        @SuppressWarnings("unchecked")
        final ValueTransformerWithKeySupplier<Integer, Integer, Iterable<String>> valueTransformerSupplier =
            mock(ValueTransformerWithKeySupplier.class);
        final KStreamFlatTransformValues<Integer, Integer, String> processorSupplier =
            new KStreamFlatTransformValues<>(valueTransformerSupplier);

        when(valueTransformerSupplier.get()).thenReturn(valueTransformer);

        final Processor<Integer, Integer, Integer, String> processor = processorSupplier.get();

        assertTrue(processor instanceof KStreamFlatTransformValuesProcessor);
    }
}
Repository: apache/kafka
File: streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamFlatTransformTest.java (Java)

Code before:
    @Test
    public void shouldAllowNullAsResultOfTransform() {
        processor.init(context);
        EasyMock.reset(transformer);

        EasyMock.expect(transformer.transform(inputKey, inputValue))
            .andReturn(null);
        replayAll();

        processor.process(new Record<>(inputKey, inputValue, 0L));

        verifyAll();
    }

    @Test
    public void shouldCloseFlatTransformProcessor() {
        transformer.close();
        replayAll();

        processor.close();

        verifyAll();
    }

    @Test
    public void shouldGetFlatTransformProcessor() {
        final TransformerSupplier<Number, Number, Iterable<KeyValue<Integer, Integer>>> transformerSupplier =
            mock(TransformerSupplier.class);
        final KStreamFlatTransform<Number, Number, Integer, Integer> processorSupplier =
            new KStreamFlatTransform<>(transformerSupplier);

        EasyMock.expect(transformerSupplier.get()).andReturn(transformer);
        replayAll();

        final Processor<Number, Number, Integer, Integer> processor = processorSupplier.get();

        verifyAll();
        assertTrue(processor instanceof KStreamFlatTransformProcessor);
    }
}


Review comment (suggestion):
nit: You could define the mock as a instance field and annotate it with `@Mock` to avoid suppressing the warning. That is a bit unfortunate in Mockito that there is no other way to set type parameters for mocks.

Code after:
        inOrder.verify(context, never()).forward(ArgumentMatchers.<Record<Integer, Integer>>any());
    }

    @Test
    public void shouldAllowNullAsResultOfTransform() {
        processor.init(context);

        when(transformer.transform(inputKey, inputValue)).thenReturn(null);

        processor.process(new Record<>(inputKey, inputValue, 0L));

        inOrder.verify(context, never()).forward(ArgumentMatchers.<Record<Integer, Integer>>any());
    }

    @Test
    public void shouldCloseFlatTransformProcessor() {
        processor.close();

        verify(transformer).close();
    }

    @Test
    public void shouldGetFlatTransformProcessor() {
        @SuppressWarnings("unchecked")
        final TransformerSupplier<Number, Number, Iterable<KeyValue<Integer, Integer>>> transformerSupplier =
            mock(TransformerSupplier.class);
        final KStreamFlatTransform<Number, Number, Integer, Integer> processorSupplier =
            new KStreamFlatTransform<>(transformerSupplier);

        when(transformerSupplier.get()).thenReturn(transformer);

        final Processor<Number, Number, Integer, Integer> processor = processorSupplier.get();

        assertTrue(processor instanceof KStreamFlatTransformProcessor);
    }
}
Repository: apache/kafka
File: connect/runtime/src/test/java/org/apache/kafka/connect/runtime/SourceTaskOffsetCommitterTest.java (Java)

Code before:
        when(taskFuture.get()).thenThrow(new CancellationException());

        committers.put(taskId, taskFuture);
        try (LogCaptureAppender logCaptureAppender = LogCaptureAppender.createAndRegister(SourceTaskOffsetCommitter.class)) {
            LogCaptureAppender.setClassLoggerToTrace(SourceTaskOffsetCommitter.class);
            committer.remove(taskId);
            assertTrue(logCaptureAppender.getEvents().stream().anyMatch(e -> e.getLevel().equals("TRACE")));
        }
        assertTrue(committers.isEmpty());
    }

    @Test
    public void testRemoveTaskAndInterrupted() throws ExecutionException, InterruptedException {
        removeSetup();
        when(taskFuture.get()).thenThrow(new InterruptedException());

        try {
            committers.put(taskId, taskFuture);
            committer.remove(taskId);
            fail("Expected ConnectException to be raised");
        } catch (ConnectException e) {
            //ignore
        }
    }

    private void removeSetup() {
        when(taskFuture.cancel(false)).thenReturn(false);
        when(taskFuture.isDone()).thenReturn(false);
        when(taskId.connector()).thenReturn("MyConnector");
        when(taskId.task()).thenReturn(1);
    }

}


Review comment (refactor):
Nit: "remove setup" kind of sounds like "remove" is a verb and we're removing something. Maybe `expectRemove`, which would also line up nicely with other utility functions in other tests?

Code after:
        when(taskFuture.get()).thenThrow(new CancellationException());

        committers.put(taskId, taskFuture);
        try (LogCaptureAppender logCaptureAppender = LogCaptureAppender.createAndRegister(SourceTaskOffsetCommitter.class)) {
            LogCaptureAppender.setClassLoggerToTrace(SourceTaskOffsetCommitter.class);
            committer.remove(taskId);
            assertTrue(logCaptureAppender.getEvents().stream().anyMatch(e -> e.getLevel().equals("TRACE")));
        }
        assertTrue(committers.isEmpty());
    }

    @Test
    public void testRemoveTaskAndInterrupted() throws ExecutionException, InterruptedException {
        expectRemove();
        when(taskFuture.get()).thenThrow(new InterruptedException());

        committers.put(taskId, taskFuture);
        assertThrows(ConnectException.class, () -> committer.remove(taskId));
    }

    private void expectRemove() {
        when(taskFuture.cancel(false)).thenReturn(false);
        when(taskFuture.isDone()).thenReturn(false);
        when(taskId.connector()).thenReturn("MyConnector");
        when(taskId.task()).thenReturn(1);
    }

}
Repository: apache/kafka
File: connect/runtime/src/test/java/org/apache/kafka/connect/runtime/SourceTaskOffsetCommitterTest.java (Java)

Code before:
    @Test
    public void testRemoveCancelledTask() throws ExecutionException, InterruptedException {
        removeSetup();
        when(taskFuture.get()).thenThrow(new CancellationException());

        committers.put(taskId, taskFuture);
        try (LogCaptureAppender logCaptureAppender = LogCaptureAppender.createAndRegister(SourceTaskOffsetCommitter.class)) {
            LogCaptureAppender.setClassLoggerToTrace(SourceTaskOffsetCommitter.class);
            committer.remove(taskId);
            assertTrue(logCaptureAppender.getEvents().stream().anyMatch(e -> e.getLevel().equals("TRACE")));
        }
        assertTrue(committers.isEmpty());
    }

    @Test
    public void testRemoveTaskAndInterrupted() throws ExecutionException, InterruptedException {
        removeSetup();
        when(taskFuture.get()).thenThrow(new InterruptedException());

        try {
            committers.put(taskId, taskFuture);
            committer.remove(taskId);
            fail("Expected ConnectException to be raised");
        } catch (ConnectException e) {
            //ignore
        }
    }

    private void removeSetup() {
        when(taskFuture.cancel(false)).thenReturn(false);
        when(taskFuture.isDone()).thenReturn(false);
        when(taskId.connector()).thenReturn("MyConnector");
        when(taskId.task()).thenReturn(1);
    }

}


Review comment (suggestion):
We can replace this with `assertThrows`:
```java
        committers.put(taskId, taskFuture);
        assertThrows(ConnectException.class, () -> committer.remove(taskId));
```

Code after:
    @Test
    public void testRemoveCancelledTask() throws ExecutionException, InterruptedException {
        expectRemove();
        when(taskFuture.get()).thenThrow(new CancellationException());

        committers.put(taskId, taskFuture);
        try (LogCaptureAppender logCaptureAppender = LogCaptureAppender.createAndRegister(SourceTaskOffsetCommitter.class)) {
            LogCaptureAppender.setClassLoggerToTrace(SourceTaskOffsetCommitter.class);
            committer.remove(taskId);
            assertTrue(logCaptureAppender.getEvents().stream().anyMatch(e -> e.getLevel().equals("TRACE")));
        }
        assertTrue(committers.isEmpty());
    }

    @Test
    public void testRemoveTaskAndInterrupted() throws ExecutionException, InterruptedException {
        expectRemove();
        when(taskFuture.get()).thenThrow(new InterruptedException());

        committers.put(taskId, taskFuture);
        assertThrows(ConnectException.class, () -> committer.remove(taskId));
    }

    private void expectRemove() {
        when(taskFuture.cancel(false)).thenReturn(false);
        when(taskFuture.isDone()).thenReturn(false);
        when(taskId.connector()).thenReturn("MyConnector");
        when(taskId.task()).thenReturn(1);
    }

}
Repository: apache/kafka
File: streams/src/main/java/org/apache/kafka/streams/state/internals/NamedCache.java (Java)

Code before:

    synchronized LRUCacheEntry last() {
        if (tail == null) {
            return null;
        }
        return tail.entry;
    }

    synchronized LRUNode head() {
        return head;
    }

    synchronized LRUNode tail() {
        return tail;
    }

    synchronized void close() {
        head = tail = null;
        listener = null;
        currentSizeBytes = 0;
        dirtyKeys.clear();
        cache.clear();
        streamsMetrics.removeAllCacheLevelSensors(Thread.currentThread().getName(), taskName, storeName);
    }

    /**
     * A simple wrapper class to implement a doubly-linked list around MemoryLRUCacheBytesEntry
     */
    static class LRUNode {
        private final Bytes key;
        private LRUCacheEntry entry;
        private LRUNode previous;
        private LRUNode next;

        LRUNode(final Bytes key, final LRUCacheEntry entry) {
            this.key = key;
            this.entry = entry;
        }

        LRUCacheEntry entry() {
            return entry;
        }

        Bytes key() {
            return key;
        }

        long size() {
            return key.get().length +
                8 + // entry
                8 + // previous

Review comment (performance):
Just for my understanding -- we know that we did `flush()` the cache already, so it's safe to just clear it?

Code after:

    synchronized LRUCacheEntry last() {
        if (tail == null) {
            return null;
        }
        return tail.entry;
    }

    synchronized LRUNode head() {
        return head;
    }

    synchronized LRUNode tail() {
        return tail;
    }

    synchronized void close() {
        head = tail = null;
        listener = null;
        currentSizeBytes = 0;
        dirtyKeys.clear();
        cache.clear();
        streamsMetrics.removeAllCacheLevelSensors(Thread.currentThread().getName(), taskName, storeName);
    }

    synchronized void clear() {
        head = tail = null;
        currentSizeBytes = 0;
        dirtyKeys.clear();
        cache.clear();
    }

    /**
     * A simple wrapper class to implement a doubly-linked list around MemoryLRUCacheBytesEntry
     */
    static class LRUNode {
        private final Bytes key;
        private LRUCacheEntry entry;
        private LRUNode previous;
        private LRUNode next;

        LRUNode(final Bytes key, final LRUCacheEntry entry) {
            this.key = key;
            this.entry = entry;
        }

        LRUCacheEntry entry() {
            return entry;
        }

        Bytes key() {
Repository: apache/kafka
File: clients/src/main/java/org/apache/kafka/clients/consumer/internals/AutoOffsetResetStrategy.java (Java)

Code before:
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;
        AutoOffsetResetStrategy that = (AutoOffsetResetStrategy) o;
        return type == that.type && Objects.equals(duration, that.duration);
    }

    @Override
    public int hashCode() {
        return Objects.hash(type, duration);
    }

    @Override
    public String toString() {
        return "AutoOffsetResetStrategy{" +
                "type=" + type +
                ", duration=" + duration +
                '}';
    }

    public static class Validator implements ConfigDef.Validator {
        @Override
        public void ensureValid(String name, Object value) {
            String strategy = (String) value;
            if (!AutoOffsetResetStrategy.isValid(strategy)) {
                throw new ConfigException(name, value, "Invalid value " + strategy + " for configuration " +
                        name + ": the value must be either 'earliest', 'latest', or 'none' or of the format 'by_duration:<PnDTnHnMn.nS.>'.");
            }
        }
    }
}


Review comment (suggestion):
```suggestion
                        name + ": the value must be either 'earliest', 'latest', 'none' or of the format 'by_duration:<PnDTnHnMn.nS.>'.");
```

Code after:
        return type == that.type && Objects.equals(duration, that.duration);
    }

    @Override
    public int hashCode() {
        return Objects.hash(type, duration);
    }

    @Override
    public String toString() {
        return "AutoOffsetResetStrategy{" +
                "type=" + type +
                (duration.map(value -> ", duration=" + value).orElse("")) +
                '}';
    }

    public static class Validator implements ConfigDef.Validator {
        @Override
        public void ensureValid(String name, Object value) {
            String offsetStrategy = (String) value;
            try {
                fromString(offsetStrategy);
            } catch (Exception e) {
                throw new ConfigException(name, value, "Invalid value `" + offsetStrategy + "` for configuration " +
                        name + ". The value must be either 'earliest', 'latest', 'none' or of the format 'by_duration:<PnDTnHnMn.nS.>'.");
            }
        }
    }
}
Repository: apache/kafka
File: core/src/main/scala/kafka/server/DelegationTokenManager.scala (Scala)

Code before:
  protected def updateCache(token: DelegationToken): Unit = {
    val hmacString = token.hmacAsBase64String
    val scramCredentialMap =  prepareScramCredentials(hmacString)
    tokenCache.updateCache(token, scramCredentialMap.asJava)
  }
  /**
   * @param hmacString
   */
  private def prepareScramCredentials(hmacString: String) : Map[String, ScramCredential] = {
    val scramCredentialMap = mutable.Map[String, ScramCredential]()

    def scramCredential(mechanism: ScramMechanism): ScramCredential = {
      new ScramFormatter(mechanism).generateCredential(hmacString, mechanism.minIterations)
    }

    for (mechanism <- ScramMechanism.values)
      scramCredentialMap(mechanism.mechanismName) = scramCredential(mechanism)

    scramCredentialMap.toMap
  }

  /**
   * @param token
   */
  def updateToken(token: DelegationToken): Unit = {
    updateCache(token)
  }

  def getDelegationToken(tokenInfo: TokenInformation): DelegationToken = {
    val hmac = createHmac(tokenInfo.tokenId, secretKey)
    new DelegationToken(tokenInfo, hmac)
  }

  /**
   *
   * @param tokenId
   */
  def removeToken(tokenId: String): Unit = {
    removeCache(tokenId)
  }

  /**
   *
   * @param tokenId
   */
  protected def removeCache(tokenId: String): Unit = {
    tokenCache.removeCache(tokenId)
  }

  def getAllTokenInformation: List[TokenInformation] = tokenCache.tokens.asScala.toList


Review comment (performance):
Could you please inline `protected def updateCache(token: DelegationToken)`?

Code after:
    scramCredentialMap.toMap
  }

  /**
   * @param token
   */
  def updateToken(token: DelegationToken): Unit = {
    val hmacString = token.hmacAsBase64String
    val scramCredentialMap = prepareScramCredentials(hmacString)
    tokenCache.updateCache(token, scramCredentialMap.asJava)
  }

  def getDelegationToken(tokenInfo: TokenInformation): DelegationToken = {
    val hmac = createHmac(tokenInfo.tokenId, secretKey)
    new DelegationToken(tokenInfo, hmac)
  }

  /**
   *
   * @param tokenId
   */
  def removeToken(tokenId: String): Unit = {
    tokenCache.removeCache(tokenId)
  }

  def getTokens(filterToken: TokenInformation => Boolean): util.List[DelegationToken] = {
    tokenCache.tokens.stream().filter(token => filterToken(token)).map(getDelegationToken).collect(Collectors.toList[DelegationToken])
  }
}