[arthas@4976]$ vmtool -x 1 --action getInstances --className kafka.log.stream.s3.network.ControllerRequestSender --express 'instances[0].requestAccumulatorMap.keySet()'
@KeySetView[
@ApiKeys[GET_KVS],
@ApiKeys[OPEN_STREAMS],
]
[arthas@4976]$ vmtool -x 3 --action getInstances --className kafka.log.stream.s3.network.ControllerRequestSender$RequestAccumulator --express 'instances[0].inflight.get()'
@Boolean[true]
[arthas@4976]$ vmtool -x 3 --action getInstances --className kafka.log.stream.s3.network.ControllerRequestSender$RequestAccumulator --express 'instances[1].inflight.get()'
@Boolean[true]
[arthas@4976]$ vmtool -x 3 --action getInstances --className kafka.log.stream.s3.network.ControllerRequestSender$RequestAccumulator --express 'instances[0]'
@RequestAccumulator[
inflight=@AtomicBoolean[
requestQueue=@LinkedBlockingQueue[
@RequestTask[
request=@[kafka.log.stream.s3.ControllerKVClient$3@6e91c450],
future=@CompletableFuture[java.util.concurrent.CompletableFuture@6c382f19[Not completed, 1 dependents]],
responseHandler=@ControllerKVClient$$Lambda$2023/0x00007f84e0805620[kafka.log.stream.s3.ControllerKVClient$$Lambda$2023/0x00007f84e0805620@27cef49f],
sendCount=@Integer[1],
],
@RequestTask[
request=@[kafka.log.stream.s3.ControllerKVClient$3@3d0daf1a],
future=@CompletableFuture[java.util.concurrent.CompletableFuture@5d5e5faf[Not completed, 1 dependents]],
responseHandler=@ControllerKVClient$$Lambda$2023/0x00007f84e0805620[kafka.log.stream.s3.ControllerKVClient$$Lambda$2023/0x00007f84e0805620@7033e3fa],
sendCount=@Integer[1],
],
@RequestTask[
request=@[kafka.log.stream.s3.ControllerKVClient$3@20099b6f],
future=@CompletableFuture[java.util.concurrent.CompletableFuture@702c850b[Not completed, 1 dependents]],
responseHandler=@ControllerKVClient$$Lambda$2023/0x00007f84e0805620[kafka.log.stream.s3.ControllerKVClient$$Lambda$2023/0x00007f84e0805620@5fa01f86],
sendCount=@Integer[1],
],
@RequestTask[
request=@[kafka.log.stream.s3.ControllerKVClient$3@30e31bbb],
future=@CompletableFuture[java.util.concurrent.CompletableFuture@5cdd3c2b[Not completed, 1 dependents]],
responseHandler=@ControllerKVClient$$Lambda$2023/0x00007f84e0805620[kafka.log.stream.s3.ControllerKVClient$$Lambda$2023/0x00007f84e0805620@484dee66],
sendCount=@Integer[1],
],
@RequestTask[
request=@[kafka.log.stream.s3.ControllerKVClient$3@17ce84c6],
future=@CompletableFuture[java.util.concurrent.CompletableFuture@3f587b47[Not completed, 1 dependents]],
responseHandler=@ControllerKVClient$$Lambda$2023/0x00007f84e0805620[kafka.log.stream.s3.ControllerKVClient$$Lambda$2023/0x00007f84e0805620@ad134e3],
sendCount=@Integer[1],
],
@RequestTask[
request=@[kafka.log.stream.s3.ControllerKVClient$3@2d48d501],
future=@CompletableFuture[java.util.concurrent.CompletableFuture@4ec39cda[Not completed, 1 dependents]],
responseHandler=@ControllerKVClient$$Lambda$2023/0x00007f84e0805620[kafka.log.stream.s3.ControllerKVClient$$Lambda$2023/0x00007f84e0805620@323d4406],
sendCount=@Integer[1],
],
@RequestTask[
request=@[kafka.log.stream.s3.ControllerKVClient$3@31c8088e],
future=@CompletableFuture[java.util.concurrent.CompletableFuture@5ac3336b[Not completed, 1 dependents]],
responseHandler=@ControllerKVClient$$Lambda$2023/0x00007f84e0805620[kafka.log.stream.s3.ControllerKVClient$$Lambda$2023/0x00007f84e0805620@83441a5],
sendCount=@Integer[1],
],
],
this$0=@ControllerRequestSender[
LOGGER=@Reload4jLoggerAdapter[
serialVersionUID=@Long[6182834493563598289],
logger=@Logger[org.apache.log4j.Logger@507129d1],
FQCN=@String[org.slf4j.impl.Reload4jLoggerAdapter],
serialVersionUID=@Long[9044267456635152283],
serialVersionUID=@Long[7535258609338176893],
name=@String[kafka.log.stream.s3.network.ControllerRequestSender],
],
MAX_RETRY_DELAY_MS=@Long[10000],
retryPolicyContext=@RetryPolicyContext[
maxRetryCount=@Integer[2147483647],
retryBaseDelayMs=@Long[500],
],
channelManager=@BrokerToControllerChannelManagerImpl[
controllerNodeProvider=@RaftControllerNodeProvider[kafka.server.RaftControllerNodeProvider@218cd729],
time=@SystemTime[org.apache.kafka.common.utils.SystemTime@6217e018],
metrics=@Metrics[org.apache.kafka.common.metrics.Metrics@4884ac96],
config=@KafkaConfig[kafka.server.KafkaConfig@95569bfb],
channelName=@String[s3stream-to-controller],
threadNamePrefix=@None$[None],
retryTimeoutMs=@Long[60000],
logContext=@LogContext[org.apache.kafka.common.utils.LogContext@5610836c],
manualMetadataUpdater=@ManualMetadataUpdater[org.apache.kafka.clients.ManualMetadataUpdater@b5aae96],
apiVersions=@ApiVersions[org.apache.kafka.clients.ApiVersions@66d74377],
requestThread=@BrokerToControllerRequestThread[Thread[BrokerToControllerChannelManager broker=12 name=s3stream-to-controller,5,main]],
logger=null,
logIdent=null,
bitmap$0=@Boolean[false],
],
retryService=@DelegatedScheduledExecutorService[
requestAccumulatorMap=@ConcurrentHashMap[
@ApiKeys[GET_KVS]:@RequestAccumulator[kafka.log.stream.s3.network.ControllerRequestSender$RequestAccumulator@573c0952],
@ApiKeys[OPEN_STREAMS]:@RequestAccumulator[kafka.log.stream.s3.network.ControllerRequestSender$RequestAccumulator@582fd280],
],
],
]
[arthas@4976]$ vmtool -x 3 --action getInstances --className kafka.log.stream.s3.network.ControllerRequestSender$RequestAccumulator --express 'instances[1]'
@RequestAccumulator[
inflight=@AtomicBoolean[
requestQueue=@LinkedBlockingQueue[
@RequestTask[
request=@[OpenStreamRequest(streamId=285, streamEpoch=7)],
future=@CompletableFuture[java.util.concurrent.CompletableFuture@6f164b56[Not completed, 1 dependents]],
responseHandler=@ControllerStreamManager$$Lambda$2029/0x00007f84e0807b80[kafka.log.stream.s3.streams.ControllerStreamManager$$Lambda$2029/0x00007f84e0807b80@1afa7c8],
sendCount=@Integer[1],
],
@RequestTask[
request=@[OpenStreamRequest(streamId=301, streamEpoch=25)],
future=@CompletableFuture[java.util.concurrent.CompletableFuture@279a8174[Not completed, 1 dependents]],
responseHandler=@ControllerStreamManager$$Lambda$2029/0x00007f84e0807b80[kafka.log.stream.s3.streams.ControllerStreamManager$$Lambda$2029/0x00007f84e0807b80@a8aa002],
sendCount=@Integer[1],
],
@RequestTask[
request=@[OpenStreamRequest(streamId=186, streamEpoch=20)],
future=@CompletableFuture[java.util.concurrent.CompletableFuture@579b16fe[Not completed, 1 dependents]],
responseHandler=@ControllerStreamManager$$Lambda$2029/0x00007f84e0807b80[kafka.log.stream.s3.streams.ControllerStreamManager$$Lambda$2029/0x00007f84e0807b80@2bf66a22],
sendCount=@Integer[1],
],
@RequestTask[
request=@[OpenStreamRequest(streamId=79, streamEpoch=12)],
future=@CompletableFuture[java.util.concurrent.CompletableFuture@5894cf09[Not completed, 1 dependents]],
responseHandler=@ControllerStreamManager$$Lambda$2029/0x00007f84e0807b80[kafka.log.stream.s3.streams.ControllerStreamManager$$Lambda$2029/0x00007f84e0807b80@741fea9],
sendCount=@Integer[1],
],
],
this$0=@ControllerRequestSender[
LOGGER=@Reload4jLoggerAdapter[
serialVersionUID=@Long[6182834493563598289],
logger=@Logger[org.apache.log4j.Logger@507129d1],
FQCN=@String[org.slf4j.impl.Reload4jLoggerAdapter],
serialVersionUID=@Long[9044267456635152283],
serialVersionUID=@Long[7535258609338176893],
name=@String[kafka.log.stream.s3.network.ControllerRequestSender],
],
MAX_RETRY_DELAY_MS=@Long[10000],
retryPolicyContext=@RetryPolicyContext[
maxRetryCount=@Integer[2147483647],
retryBaseDelayMs=@Long[500],
],
channelManager=@BrokerToControllerChannelManagerImpl[
controllerNodeProvider=@RaftControllerNodeProvider[kafka.server.RaftControllerNodeProvider@218cd729],
time=@SystemTime[org.apache.kafka.common.utils.SystemTime@6217e018],
metrics=@Metrics[org.apache.kafka.common.metrics.Metrics@4884ac96],
config=@KafkaConfig[kafka.server.KafkaConfig@95569bfb],
channelName=@String[s3stream-to-controller],
threadNamePrefix=@None$[None],
retryTimeoutMs=@Long[60000],
logContext=@LogContext[org.apache.kafka.common.utils.LogContext@5610836c],
manualMetadataUpdater=@ManualMetadataUpdater[org.apache.kafka.clients.ManualMetadataUpdater@b5aae96],
apiVersions=@ApiVersions[org.apache.kafka.clients.ApiVersions@66d74377],
requestThread=@BrokerToControllerRequestThread[Thread[BrokerToControllerChannelManager broker=12 name=s3stream-to-controller,5,main]],
logger=null,
logIdent=null,
bitmap$0=@Boolean[false],
],
retryService=@DelegatedScheduledExecutorService[
requestAccumulatorMap=@ConcurrentHashMap[
@ApiKeys[GET_KVS]:@RequestAccumulator[kafka.log.stream.s3.network.ControllerRequestSender$RequestAccumulator@573c0952],
@ApiKeys[OPEN_STREAMS]:@RequestAccumulator[kafka.log.stream.s3.network.ControllerRequestSender$RequestAccumulator@582fd280],
],
],
]
