Fix broken microbenchmark of processor module (#140)

* Fix broken microbenchmark of processor module

* Ensure microbench isn't broken in CI

.github/workflows/ci.yml

@@ -22,4 +22,4 @@ jobs:
     - name: Execute test
       uses: eskatos/gradle-command-action@v1
       with:
-        arguments: build integrationTest
+        arguments: build jmhJar integrationTest

build.gradle

@@ -2,6 +2,7 @@ import org.gradle.api.tasks.testing.logging.TestExceptionFormat
 import org.gradle.api.tasks.testing.logging.TestLogEvent
 
 plugins {
+    id "me.champeau.jmh" version "0.6.6" apply false
     id 'com.github.johnrengelman.shadow' version '7.1.0' apply false
     id 'io.franzbecker.gradle-lombok' version '5.0.0' apply false
 }
@@ -13,6 +14,11 @@ subprojects {
     apply plugin: 'java-test-fixtures'
     apply plugin: 'maven-publish'
     apply plugin: 'signing'
+    // !!!Do not reverse order of these two plugins (jmh and shadow) loading!!!
+    // jmh plugin changes its behavior by whether the shadow plugin is loaded or not, and with shadow plugin
+    // it does not work well by failing to generate an jmh JAR to be executed.
+    // https://github.com/melix/jmh-gradle-plugin/blob/c17c89c8fed6d655678aac16ecd5b683ea7fc8b5/src/main/groovy/me/champeau/jmh/JMHPlugin.groovy#L77
+    apply plugin: 'me.champeau.jmh'
     apply plugin: 'com.github.johnrengelman.shadow'
     apply plugin: 'io.franzbecker.gradle-lombok'
 

processor/build.gradle

@@ -1,7 +1,3 @@
-plugins {
-    id "me.champeau.jmh" version "0.6.6"
-}
-
 dependencies {
     api project(":common")
     api project(":client")

processor/src/jmh/java/com/linecorp/decaton/processor/runtime/OutOfOrderCommitControlBenchmark.java

@@ -35,6 +35,10 @@
 import org.openjdk.jmh.annotations.Warmup;
 
 import com.linecorp.decaton.processor.DeferredCompletion;
+import com.linecorp.decaton.processor.metrics.Metrics;
+import com.linecorp.decaton.processor.metrics.internal.AvailableTags;
+import com.linecorp.decaton.processor.runtime.internal.OffsetState;
+import com.linecorp.decaton.processor.runtime.internal.OffsetStateReaper;
 import com.linecorp.decaton.processor.runtime.internal.OutOfOrderCommitControl;
 
 /**
@@ -169,7 +173,11 @@ public void outOfOrderCommitControlV2(BmStateV2 state) throws InterruptedExcepti
     public static class BmStateV3 extends BmState<OutOfOrderCommitControl> {
         @Override
         OutOfOrderCommitControl createCommitControl() {
-            return new OutOfOrderCommitControl(topicPartition, CAPACITY);
+            OffsetStateReaper reaper = new OffsetStateReaper(
+                    Property.ofStatic(ProcessorProperties.CONFIG_DEFERRED_COMPLETE_TIMEOUT_MS),
+                    Metrics.withTags("subscription", "subsc", "topic", "topic", "partition", "1")
+                            .new CommitControlMetrics());
+            return new OutOfOrderCommitControl(topicPartition, CAPACITY, reaper);
         }
     }
 
@@ -185,9 +193,9 @@ public void outOfOrderCommitControlV3(BmStateV3 state) throws InterruptedExcepti
                     break;
                 }
                 noProgress = false;
-                DeferredCompletion completion = control.reportFetchedOffset(offset);
+                OffsetState offsetState = control.reportFetchedOffset(offset);
 
-                state.workers.execute(completion::complete);
+                state.workers.execute(offsetState.completion()::complete);
             }
             if (noProgress) {
                 Thread.yield();

processor/src/test/resources/logback.xml

@@ -0,0 +1,26 @@
+<!--
+  ~ Copyright 2021 LINE Corporation
+  ~
+  ~ LINE Corporation licenses this file to you under the Apache License,
+  ~ version 2.0 (the "License"); you may not use this file except in compliance
+  ~ with the License. You may obtain a copy of the License at:
+  ~
+  ~   https://www.apache.org/licenses/LICENSE-2.0
+  ~
+  ~ Unless required by applicable law or agreed to in writing, software
+  ~ distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+  ~ WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+  ~ License for the specific language governing permissions and limitations
+  ~ under the License.
+  -->
+
+<configuration>
+  <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+    <encoder>
+      <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
+    </encoder>
+  </appender>
+  <root level="ERROR">
+    <appender-ref ref="STDOUT" />
+  </root>
+</configuration>