Use TopicPartition.topic for metrics (#235)

* Use TopicPartition.topic for metrics

* Use variable and rename SubscriptionScope.topic to originTopic

* Use topicPartition.topic() for ThreadUtilizationMetrics

processor/src/main/java/com/linecorp/decaton/processor/runtime/ProcessorSubscription.java

@@ -218,7 +218,7 @@ private void updateState(SubscriptionStateListener.State newState) {
 
     private Set<String> subscribeTopics() {
         return Stream.concat(
-                Stream.of(Optional.of(scope.topic()), scope.retryTopic())
+                Stream.of(Optional.of(scope.originTopic()), scope.retryTopic())
                       .filter(Optional::isPresent)
                       .map(Optional::get),
                 scope.shapingTopics().stream()).collect(Collectors.toSet());

processor/src/main/java/com/linecorp/decaton/processor/runtime/SubscriptionBuilder.java

@@ -322,7 +322,7 @@ private QuotaApplier quotaApplier(SubscriptionScope scope) {
                                 (properties, new ByteArraySerializer(), new ByteArraySerializer()));
         return new QuotaApplierImpl(
                 producerSupplier.apply(producerConfig),
-                perKeyQuotaConfig.callbackSupplier().apply(scope.topic()),
+                perKeyQuotaConfig.callbackSupplier().apply(scope.originTopic()),
                 scope);
     }
 }

processor/src/main/java/com/linecorp/decaton/processor/runtime/internal/AbstractSubPartitions.java

@@ -29,6 +29,7 @@
 import com.linecorp.decaton.processor.runtime.Property;
 
 import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.common.TopicPartition;
 
 /**
  * This class is responsible for following portions:
@@ -57,10 +58,11 @@ protected AbstractSubPartitions(PartitionScope scope, Processors<?> processors)
         shutdownTimeoutMillis = scope.props().get(
                 ProcessorProperties.CONFIG_PROCESSOR_THREADS_TERMINATION_TIMEOUT_MS);
         rateLimiter = new DynamicRateLimiter(rateProperty(scope));
+        TopicPartition topicPartition = scope.topicPartition();
         Metrics metrics = Metrics.withTags(
                 "subscription", scope.subscriptionId(),
-                "topic", scope.topic(),
-                "partition", String.valueOf(scope.topicPartition().partition()));
+                "topic", topicPartition.topic(),
+                "partition", String.valueOf(topicPartition.partition()));
         taskMetrics = metrics.new TaskMetrics();
         schedulerMetrics = metrics.new SchedulerMetrics();
     }

processor/src/main/java/com/linecorp/decaton/processor/runtime/internal/PartitionContext.java

@@ -103,7 +103,7 @@ private static SubPartitions createSubPartitions(PartitionScope scope, Processor
                 scope.props().get(ProcessorProperties.CONFIG_DEFERRED_COMPLETE_TIMEOUT_MS),
                 metricsCtor.new CommitControlMetrics());
         commitControl = new OutOfOrderCommitControl(scope.topicPartition(), capacity, offsetStateReaper);
-        if (scope.perKeyQuotaConfig().isPresent() && scope.topic().equals(scope.topicPartition().topic())) {
+        if (scope.perKeyQuotaConfig().isPresent() && scope.originTopic().equals(scope.topicPartition().topic())) {
             perKeyQuotaManager = PerKeyQuotaManager.create(scope);
         } else {
             perKeyQuotaManager = null;

processor/src/main/java/com/linecorp/decaton/processor/runtime/internal/PartitionScope.java

@@ -29,7 +29,7 @@ public class PartitionScope extends SubscriptionScope {
     private final TopicPartition topicPartition;
 
     PartitionScope(SubscriptionScope parent, TopicPartition topicPartition) {
-        super(parent.subscriptionId(), parent.topic(), parent.subPartitionRuntime(),
+        super(parent.subscriptionId(), parent.originTopic(), parent.subPartitionRuntime(),
               parent.retryConfig(), parent.perKeyQuotaConfig(), parent.props(),
               parent.tracingProvider(), parent.maxPollRecords(), parent.subPartitionerSupplier());
         this.topicPartition = topicPartition;

processor/src/main/java/com/linecorp/decaton/processor/runtime/internal/SubscriptionScope.java

@@ -38,7 +38,7 @@
 @Getter
 public class SubscriptionScope {
     private final String subscriptionId;
-    private final String topic;
+    private final String originTopic;
     private final SubPartitionRuntime subPartitionRuntime;
     private final Optional<RetryConfig> retryConfig;
     private final Optional<PerKeyQuotaConfig> perKeyQuotaConfig;
@@ -48,11 +48,11 @@ public class SubscriptionScope {
     private final SubPartitionerSupplier subPartitionerSupplier;
 
     public Optional<String> retryTopic() {
-        return retryConfig.map(conf -> conf.retryTopicOrDefault(topic));
+        return retryConfig.map(conf -> conf.retryTopicOrDefault(originTopic));
     }
 
     public Set<String> shapingTopics() {
-        return perKeyQuotaConfig.map(conf -> conf.shapingTopicsSupplier().apply(topic))
+        return perKeyQuotaConfig.map(conf -> conf.shapingTopicsSupplier().apply(originTopic))
                                 .orElse(Collections.emptySet());
     }
 

processor/src/main/java/com/linecorp/decaton/processor/runtime/internal/ThreadPoolSubPartitions.java

@@ -32,6 +32,7 @@
 
 import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.common.TopicPartition;
 
 @Slf4j
 public class ThreadPoolSubPartitions extends AbstractSubPartitions {
@@ -67,10 +68,11 @@ public void addTask(TaskRequest request) {
         SubPartition subPartition = subPartitions[threadId];
         if (subPartition == null) {
             ThreadScope threadScope = new ThreadScope(scope, threadId);
+            TopicPartition topicPartition = threadScope.topicPartition();
             ThreadUtilizationMetrics metrics =
                     Metrics.withTags("subscription", threadScope.subscriptionId(),
-                                     "topic", threadScope.topic(),
-                                     "partition", String.valueOf(threadScope.topicPartition().partition()),
+                                     "topic", topicPartition.topic(),
+                                     "partition", String.valueOf(topicPartition.partition()),
                                      "subpartition", String.valueOf(threadId))
                             .new ThreadUtilizationMetrics();
             ExecutorService executor = createExecutorService(threadScope, metrics);

processor/src/test/java/com/linecorp/decaton/processor/runtime/ProcessorSubscriptionTest.java

@@ -138,7 +138,7 @@ private static ProcessorSubscription subscription(Consumer<byte[], byte[]> consu
                                                       DecatonProcessor<String> processor) {
         SubscriptionScope scope = scope(tp.topic(), 0L);
         ProcessorsBuilder<String> builder =
-                ProcessorsBuilder.consuming(scope.topic(),
+                ProcessorsBuilder.consuming(scope.originTopic(),
                                             (byte[] bytes) -> new DecatonTask<>(
                                                     TaskMetadata.builder().build(),
                                                     new String(bytes), bytes));
@@ -276,7 +276,7 @@ public synchronized ConsumerRecords<byte[], byte[]> poll(Duration timeout) {
                 scope,
                 consumer,
                 NoopQuotaApplier.INSTANCE,
-                ProcessorsBuilder.consuming(scope.topic(),
+                ProcessorsBuilder.consuming(scope.originTopic(),
                                             (byte[] bytes) -> new DecatonTask<>(
                                                     TaskMetadata.builder().build(), "dummy", bytes))
                                  .thenProcess(processor)
@@ -350,7 +350,7 @@ public synchronized void commitSync(Map<TopicPartition, OffsetAndMetadata> offse
                 scope,
                 consumer,
                 NoopQuotaApplier.INSTANCE,
-                ProcessorsBuilder.consuming(scope.topic(),
+                ProcessorsBuilder.consuming(scope.originTopic(),
                                             (byte[] bytes) -> new DecatonTask<>(
                                                     TaskMetadata.builder().build(), "dummy", bytes))
                                  .thenProcess((ctx, task) -> {