kkonstantine commented on a change in pull request #10053:
URL: https://github.com/apache/kafka/pull/10053#discussion_r570464187
##########
File path:
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java
##########
@@ -236,7 +236,7 @@ public DistributedHerder(DistributedConfig config,
this.herderExecutor = new ThreadPoolExecutor(1, 1, 0L,
TimeUnit.MILLISECONDS,
- new LinkedBlockingDeque<Runnable>(1),
+ new LinkedBlockingDeque<>(1),
Review comment:
same nit as above
##########
File path:
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/SourceTaskOffsetCommitter.java
##########
@@ -63,7 +63,7 @@
public SourceTaskOffsetCommitter(WorkerConfig config) {
this(config,
Executors.newSingleThreadScheduledExecutor(ThreadUtils.createThreadFactory(
SourceTaskOffsetCommitter.class.getSimpleName() + "-%d",
false)),
- new ConcurrentHashMap<ConnectorTaskId, ScheduledFuture<?>>());
+ new ConcurrentHashMap<>());
Review comment:
nit: format got unaligned. Please check the suggestion fixes it
```suggestion
new ConcurrentHashMap<>());
```
##########
File path:
connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStoreTest.java
##########
@@ -269,8 +269,10 @@ public void testGetSetNull() throws Exception {
final Capture<Callback<Void>> secondGetReadToEndCallback =
EasyMock.newCapture();
storeLog.readToEnd(EasyMock.capture(secondGetReadToEndCallback));
PowerMock.expectLastCall().andAnswer(() -> {
- capturedConsumedCallback.getValue().onCompletion(null, new
ConsumerRecord<>(TOPIC, 0, 0, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, (byte[])
null, TP0_VALUE.array()));
- capturedConsumedCallback.getValue().onCompletion(null, new
ConsumerRecord<>(TOPIC, 1, 0, 0L, TimestampType.CREATE_TIME, 0L, 0, 0,
TP1_KEY.array(), (byte[]) null));
+ capturedConsumedCallback.getValue().onCompletion(null, new
ConsumerRecord<>(TOPIC, 0, 0, 0L, TimestampType.CREATE_TIME, 0L, 0, 0,
+ null, TP0_VALUE.array()));
+ capturedConsumedCallback.getValue().onCompletion(null, new
ConsumerRecord<>(TOPIC, 1, 0, 0L, TimestampType.CREATE_TIME, 0L, 0, 0,
TP1_KEY.array(),
+ null));
Review comment:
```suggestion
capturedConsumedCallback.getValue().onCompletion(null, new
ConsumerRecord<>(TOPIC, 1, 0, 0L, TimestampType.CREATE_TIME, 0L, 0, 0,
TP1_KEY.array(), null));
```
##########
File path:
connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaOffsetBackingStoreTest.java
##########
@@ -269,8 +269,10 @@ public void testGetSetNull() throws Exception {
final Capture<Callback<Void>> secondGetReadToEndCallback =
EasyMock.newCapture();
storeLog.readToEnd(EasyMock.capture(secondGetReadToEndCallback));
PowerMock.expectLastCall().andAnswer(() -> {
- capturedConsumedCallback.getValue().onCompletion(null, new
ConsumerRecord<>(TOPIC, 0, 0, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, (byte[])
null, TP0_VALUE.array()));
- capturedConsumedCallback.getValue().onCompletion(null, new
ConsumerRecord<>(TOPIC, 1, 0, 0L, TimestampType.CREATE_TIME, 0L, 0, 0,
TP1_KEY.array(), (byte[]) null));
+ capturedConsumedCallback.getValue().onCompletion(null, new
ConsumerRecord<>(TOPIC, 0, 0, 0L, TimestampType.CREATE_TIME, 0L, 0, 0,
+ null, TP0_VALUE.array()));
Review comment:
```suggestion
capturedConsumedCallback.getValue().onCompletion(null, new
ConsumerRecord<>(TOPIC, 0, 0, 0L, TimestampType.CREATE_TIME, 0L, 0, 0, null,
TP0_VALUE.array()));
```
##########
File path:
connect/runtime/src/main/java/org/apache/kafka/connect/connector/policy/AbstractConnectorClientConfigOverridePolicy.java
##########
@@ -39,7 +39,8 @@ public void close() throws Exception {
protected ConfigValue configValue(Map.Entry<String, Object> configEntry) {
ConfigValue configValue =
- new ConfigValue(configEntry.getKey(), configEntry.getValue(), new
ArrayList<>(), new ArrayList<String>());
+ new ConfigValue(configEntry.getKey(), configEntry.getValue(), new
ArrayList<>(),
+ new ArrayList<>());
Review comment:
In this project width can be longer. For old lines let's keep it like
that. New is fine to format based on a 100 char width (I believe) but again not
required currently.
```suggestion
new ConfigValue(configEntry.getKey(), configEntry.getValue(),
new ArrayList<>(), new ArrayList<>());
```
##########
File path:
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorsResource.java
##########
@@ -390,12 +390,14 @@ private void checkAndPutConnectorConfigName(String
connectorName, Map<String, St
private <T> T completeOrForwardRequest(FutureCallback<T> cb, String path,
String method, HttpHeaders headers, Object body,
TypeReference<T> resultType,
Boolean forward) throws Throwable {
- return completeOrForwardRequest(cb, path, method, headers, body,
resultType, new IdentityTranslator<T>(), forward);
+ return completeOrForwardRequest(cb, path, method, headers, body,
resultType,
+ new IdentityTranslator<>(), forward);
}
private <T> T completeOrForwardRequest(FutureCallback<T> cb, String path,
String method, HttpHeaders headers,
Object body, Boolean forward)
throws Throwable {
- return completeOrForwardRequest(cb, path, method, headers, body, null,
new IdentityTranslator<T>(), forward);
+ return completeOrForwardRequest(cb, path, method, headers, body, null,
+ new IdentityTranslator<>(), forward);
Review comment:
```suggestion
return completeOrForwardRequest(cb, path, method, headers, body,
null, new IdentityTranslator<>(), forward);
```
##########
File path:
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorsResource.java
##########
@@ -390,12 +390,14 @@ private void checkAndPutConnectorConfigName(String
connectorName, Map<String, St
private <T> T completeOrForwardRequest(FutureCallback<T> cb, String path,
String method, HttpHeaders headers, Object body,
TypeReference<T> resultType,
Boolean forward) throws Throwable {
- return completeOrForwardRequest(cb, path, method, headers, body,
resultType, new IdentityTranslator<T>(), forward);
+ return completeOrForwardRequest(cb, path, method, headers, body,
resultType,
+ new IdentityTranslator<>(), forward);
Review comment:
```suggestion
return completeOrForwardRequest(cb, path, method, headers, body,
resultType, new IdentityTranslator<>(), forward);
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
