abhijeetk88 commented on code in PR #14127:
URL: https://github.com/apache/kafka/pull/14127#discussion_r1299368244
##########
storage/src/main/java/org/apache/kafka/server/log/remote/metadata/storage/ConsumerTask.java:
##########
@@ -64,302 +65,387 @@
class ConsumerTask implements Runnable, Closeable {
private static final Logger log =
LoggerFactory.getLogger(ConsumerTask.class);
- private static final long POLL_INTERVAL_MS = 100L;
+ static long pollIntervalMs = 100L;
private final RemoteLogMetadataSerde serde = new RemoteLogMetadataSerde();
- private final KafkaConsumer<byte[], byte[]> consumer;
- private final String metadataTopicName;
+ private final Consumer<byte[], byte[]> consumer;
private final RemotePartitionMetadataEventHandler
remotePartitionMetadataEventHandler;
private final RemoteLogMetadataTopicPartitioner topicPartitioner;
- private final Time time;
+ private final Time time = new SystemTime();
+ // TODO - Update comments below
// It indicates whether the closing process has been started or not. If it
is set as true,
// consumer will stop consuming messages, and it will not allow partition
assignments to be updated.
- private volatile boolean closing = false;
-
+ private volatile boolean isClosed = false;
// It indicates whether the consumer needs to assign the partitions or
not. This is set when it is
// determined that the consumer needs to be assigned with the updated
partitions.
- private volatile boolean assignPartitions = false;
+ private volatile boolean isAssignmentChanged = true;
Review Comment:
Added a note in the comment
--
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.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
