Merge branch 'main' into enhance-per-tool

app/src/main/java/org/astraea/app/performance/Performance.java

@@ -98,25 +98,33 @@ public static List<String> execute(final Argument param)
                 param.consumers,
                 (clientId, listener) ->
                     Consumer.forTopics(new HashSet<>(param.topics))
-                        .bootstrapServers(param.bootstrapServers())
-                        .groupId(param.groupId)
                         .configs(param.configs())
-                        .isolation(param.isolation())
+                        .config(
+                            ConsumerConfigs.ISOLATION_LEVEL_CONFIG,
+                            param.transactionSize > 1
+                                ? ConsumerConfigs.ISOLATION_LEVEL_COMMITTED
+                                : ConsumerConfigs.ISOLATION_LEVEL_UNCOMMITTED)
+                        .bootstrapServers(param.bootstrapServers())
+                        .config(ConsumerConfigs.GROUP_ID_CONFIG, param.groupId)
                         .seek(latestOffsets)
                         .consumerRebalanceListener(listener)
-                        .clientId(clientId)
+                        .config(ConsumerConfigs.CLIENT_ID_CONFIG, clientId)
                         .build())
             : ConsumerThread.create(
                 param.consumers,
                 (clientId, listener) ->
                     Consumer.forTopics(param.pattern)
-                        .bootstrapServers(param.bootstrapServers())
-                        .groupId(param.groupId)
                         .configs(param.configs())
-                        .isolation(param.isolation())
+                        .config(
+                            ConsumerConfigs.ISOLATION_LEVEL_CONFIG,
+                            param.transactionSize > 1
+                                ? ConsumerConfigs.ISOLATION_LEVEL_COMMITTED
+                                : ConsumerConfigs.ISOLATION_LEVEL_UNCOMMITTED)
+                        .bootstrapServers(param.bootstrapServers())
+                        .config(ConsumerConfigs.GROUP_ID_CONFIG, param.groupId)
                         .seek(latestOffsets)
                         .consumerRebalanceListener(listener)
-                        .clientId(clientId)
+                        .config(ConsumerConfigs.CLIENT_ID_CONFIG, clientId)
                         .build());
 
     System.out.println("creating tracker");