[GitHub] [kafka] mjsax commented on a change in pull request #8876: KAFKA-10167: use the admin client to read end-offset

[GitBox]

mjsax commented on a change in pull request #8876:
URL: https://github.com/apache/kafka/pull/8876#discussion_r440535836



##########
File path: 
clients/src/test/java/org/apache/kafka/clients/admin/MockAdminClient.java
##########
@@ -834,6 +855,13 @@ public AlterClientQuotasResult 
alterClientQuotas(Collection<ClientQuotaAlteratio
     @Override
     synchronized public void close(Duration timeout) {}
 
+    public synchronized void updateBeginningOffsets(Map<TopicPartition, Long> 
newOffsets) {
+        beginningOffsets.putAll(newOffsets);
+    }
+    public synchronized void updateEndOffsets(final Map<TopicPartition, Long> 
newOffsets) {

Review comment:
       nit: empty line.

##########
File path: 
streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java
##########
@@ -564,8 +576,15 @@ private void restoreChangelog(final ChangelogMetadata 
changelogMetadata) {
             return Collections.emptyMap();
 
         try {
-            return restoreConsumer.endOffsets(partitions);
-        } catch (final TimeoutException e) {
+            if (adminClient != null) {

Review comment:
       Why do we need this distinction? Seems we set `adminClient` in any case 
and it should never be `null`?

##########
File path: 
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
##########
@@ -273,13 +273,13 @@ public boolean isRunning() {
     private volatile ThreadMetadata threadMetadata;
     private StreamThread.StateListener stateListener;
 
-    private final Admin adminClient;
     private final ChangelogReader changelogReader;
 
     // package-private for testing
     final ConsumerRebalanceListener rebalanceListener;
     final Consumer<byte[], byte[]> mainConsumer;
     final Consumer<byte[], byte[]> restoreConsumer;
+    final Admin adminClient;

Review comment:
       Why is `adminClient` not final any longer? We still pass it into the 
`StreamThread` constructor.




----------------------------------------------------------------
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:
us...@infra.apache.org