> Task :app:compileJava > Task :app:processResources UP-TO-DATE > Task :app:classes > Task :app:run INFO 10:22:27.124[main] StreamsConfig - StreamsConfig values: acceptable.recovery.lag = 10000 application.id = forwarding-stream application.server = bootstrap.servers = [localhost:9092] buffered.records.per.partition = 1000 built.in.metrics.version = latest cache.max.bytes.buffering = 10485760 client.id = commit.interval.ms = 10000 connections.max.idle.ms = 540000 default.deserialization.exception.handler = class org.apache.kafka.streams.errors.LogAndFailExceptionHandler default.key.serde = class org.apache.kafka.common.serialization.Serdes$ByteArraySerde default.production.exception.handler = class org.apache.kafka.streams.errors.DefaultProductionExceptionHandler default.timestamp.extractor = class org.apache.kafka.streams.processor.FailOnInvalidTimestamp default.value.serde = class org.apache.kafka.common.serialization.Serdes$ByteArraySerde default.windowed.key.serde.inner = null default.windowed.value.serde.inner = null max.task.idle.ms = 0 max.warmup.replicas = 2 metadata.max.age.ms = 300000 metric.reporters = [] metrics.num.samples = 2 metrics.recording.level = INFO metrics.sample.window.ms = 30000 num.standby.replicas = 0 num.stream.threads = 1 partition.grouper = class org.apache.kafka.streams.processor.DefaultPartitionGrouper poll.ms = 100 probing.rebalance.interval.ms = 600000 processing.guarantee = exactly_once receive.buffer.bytes = 32768 reconnect.backoff.max.ms = 1000 reconnect.backoff.ms = 50 replication.factor = 1 request.timeout.ms = 40000 retries = 0 retry.backoff.ms = 100 rocksdb.config.setter = null security.protocol = PLAINTEXT send.buffer.bytes = 131072 state.cleanup.delay.ms = 600000 state.dir = /tmp/kafka-streams task.timeout.ms = 300000 topology.optimization = none upgrade.from = null window.size.ms = null windowstore.changelog.additional.retention.ms = 86400000 INFO 10:22:27.148[main] AdminClientConfig - AdminClientConfig values: bootstrap.servers = [localhost:9092] client.dns.lookup = use_all_dns_ips client.id = forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-admin connections.max.idle.ms = 300000 default.api.timeout.ms = 60000 metadata.max.age.ms = 300000 metric.reporters = [] metrics.num.samples = 2 metrics.recording.level = INFO metrics.sample.window.ms = 30000 receive.buffer.bytes = 65536 reconnect.backoff.max.ms = 1000 reconnect.backoff.ms = 50 request.timeout.ms = 30000 retries = 2147483647 retry.backoff.ms = 100 sasl.client.callback.handler.class = null sasl.jaas.config = null sasl.kerberos.kinit.cmd = /usr/bin/kinit sasl.kerberos.min.time.before.relogin = 60000 sasl.kerberos.service.name = null sasl.kerberos.ticket.renew.jitter = 0.05 sasl.kerberos.ticket.renew.window.factor = 0.8 sasl.login.callback.handler.class = null sasl.login.class = null sasl.login.refresh.buffer.seconds = 300 sasl.login.refresh.min.period.seconds = 60 sasl.login.refresh.window.factor = 0.8 sasl.login.refresh.window.jitter = 0.05 sasl.mechanism = GSSAPI security.protocol = PLAINTEXT security.providers = null send.buffer.bytes = 131072 socket.connection.setup.timeout.max.ms = 30000 socket.connection.setup.timeout.ms = 10000 ssl.cipher.suites = null ssl.enabled.protocols = [TLSv1.2, TLSv1.3] ssl.endpoint.identification.algorithm = https ssl.engine.factory.class = null ssl.key.password = null ssl.keymanager.algorithm = SunX509 ssl.keystore.certificate.chain = null ssl.keystore.key = null ssl.keystore.location = null ssl.keystore.password = null ssl.keystore.type = JKS ssl.protocol = TLSv1.3 ssl.provider = null ssl.secure.random.implementation = null ssl.trustmanager.algorithm = PKIX ssl.truststore.certificates = null ssl.truststore.location = null ssl.truststore.password = null ssl.truststore.type = JKS INFO 10:22:27.325[main] AppInfoParser - Kafka version: 2.8.0 INFO 10:22:27.326[main] AppInfoParser - Kafka commitId: ebb1d6e21cc92130 INFO 10:22:27.326[main] AppInfoParser - Kafka startTimeMs: 1622535747324 INFO 10:22:27.330[main] KafkaStreams - stream-client [forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c] Kafka Streams version: 2.8.0 INFO 10:22:27.330[main] KafkaStreams - stream-client [forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c] Kafka Streams commit ID: ebb1d6e21cc92130 INFO 10:22:27.339[main] StreamThread - stream-thread [forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] Creating restore consumer client INFO 10:22:27.343[main] ConsumerConfig - ConsumerConfig values: allow.auto.create.topics = true auto.commit.interval.ms = 5000 auto.offset.reset = none bootstrap.servers = [localhost:9092] check.crcs = true client.dns.lookup = use_all_dns_ips client.id = forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-restore-consumer client.rack = connections.max.idle.ms = 540000 default.api.timeout.ms = 60000 enable.auto.commit = false exclude.internal.topics = true fetch.max.bytes = 52428800 fetch.max.wait.ms = 500 fetch.min.bytes = 1 group.id = null group.instance.id = null heartbeat.interval.ms = 3000 interceptor.classes = [] internal.leave.group.on.close = false internal.throw.on.fetch.stable.offset.unsupported = false isolation.level = read_committed key.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer max.partition.fetch.bytes = 1048576 max.poll.interval.ms = 300000 max.poll.records = 1000 metadata.max.age.ms = 300000 metric.reporters = [] metrics.num.samples = 2 metrics.recording.level = INFO metrics.sample.window.ms = 30000 partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor] receive.buffer.bytes = 65536 reconnect.backoff.max.ms = 1000 reconnect.backoff.ms = 50 request.timeout.ms = 30000 retry.backoff.ms = 100 sasl.client.callback.handler.class = null sasl.jaas.config = null sasl.kerberos.kinit.cmd = /usr/bin/kinit sasl.kerberos.min.time.before.relogin = 60000 sasl.kerberos.service.name = null sasl.kerberos.ticket.renew.jitter = 0.05 sasl.kerberos.ticket.renew.window.factor = 0.8 sasl.login.callback.handler.class = null sasl.login.class = null sasl.login.refresh.buffer.seconds = 300 sasl.login.refresh.min.period.seconds = 60 sasl.login.refresh.window.factor = 0.8 sasl.login.refresh.window.jitter = 0.05 sasl.mechanism = GSSAPI security.protocol = PLAINTEXT security.providers = null send.buffer.bytes = 131072 session.timeout.ms = 10000 socket.connection.setup.timeout.max.ms = 30000 socket.connection.setup.timeout.ms = 10000 ssl.cipher.suites = null ssl.enabled.protocols = [TLSv1.2, TLSv1.3] ssl.endpoint.identification.algorithm = https ssl.engine.factory.class = null ssl.key.password = null ssl.keymanager.algorithm = SunX509 ssl.keystore.certificate.chain = null ssl.keystore.key = null ssl.keystore.location = null ssl.keystore.password = null ssl.keystore.type = JKS ssl.protocol = TLSv1.3 ssl.provider = null ssl.secure.random.implementation = null ssl.trustmanager.algorithm = PKIX ssl.truststore.certificates = null ssl.truststore.location = null ssl.truststore.password = null ssl.truststore.type = JKS value.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer INFO 10:22:27.371[main] AppInfoParser - Kafka version: 2.8.0 INFO 10:22:27.371[main] AppInfoParser - Kafka commitId: ebb1d6e21cc92130 INFO 10:22:27.371[main] AppInfoParser - Kafka startTimeMs: 1622535747371 INFO 10:22:27.380[main] StreamThread - stream-thread [forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] Creating consumer client INFO 10:22:27.383[main] ConsumerConfig - ConsumerConfig values: allow.auto.create.topics = false auto.commit.interval.ms = 5000 auto.offset.reset = earliest bootstrap.servers = [localhost:9092] check.crcs = true client.dns.lookup = use_all_dns_ips client.id = forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-consumer client.rack = connections.max.idle.ms = 540000 default.api.timeout.ms = 60000 enable.auto.commit = false exclude.internal.topics = true fetch.max.bytes = 52428800 fetch.max.wait.ms = 500 fetch.min.bytes = 1 group.id = forwarding-stream group.instance.id = null heartbeat.interval.ms = 3000 interceptor.classes = [] internal.leave.group.on.close = false internal.throw.on.fetch.stable.offset.unsupported = false isolation.level = read_committed key.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer max.partition.fetch.bytes = 1048576 max.poll.interval.ms = 300000 max.poll.records = 1000 metadata.max.age.ms = 300000 metric.reporters = [] metrics.num.samples = 2 metrics.recording.level = INFO metrics.sample.window.ms = 30000 partition.assignment.strategy = [org.apache.kafka.streams.processor.internals.StreamsPartitionAssignor] receive.buffer.bytes = 65536 reconnect.backoff.max.ms = 1000 reconnect.backoff.ms = 50 request.timeout.ms = 30000 retry.backoff.ms = 100 sasl.client.callback.handler.class = null sasl.jaas.config = null sasl.kerberos.kinit.cmd = /usr/bin/kinit sasl.kerberos.min.time.before.relogin = 60000 sasl.kerberos.service.name = null sasl.kerberos.ticket.renew.jitter = 0.05 sasl.kerberos.ticket.renew.window.factor = 0.8 sasl.login.callback.handler.class = null sasl.login.class = null sasl.login.refresh.buffer.seconds = 300 sasl.login.refresh.min.period.seconds = 60 sasl.login.refresh.window.factor = 0.8 sasl.login.refresh.window.jitter = 0.05 sasl.mechanism = GSSAPI security.protocol = PLAINTEXT security.providers = null send.buffer.bytes = 131072 session.timeout.ms = 10000 socket.connection.setup.timeout.max.ms = 30000 socket.connection.setup.timeout.ms = 10000 ssl.cipher.suites = null ssl.enabled.protocols = [TLSv1.2, TLSv1.3] ssl.endpoint.identification.algorithm = https ssl.engine.factory.class = null ssl.key.password = null ssl.keymanager.algorithm = SunX509 ssl.keystore.certificate.chain = null ssl.keystore.key = null ssl.keystore.location = null ssl.keystore.password = null ssl.keystore.type = JKS ssl.protocol = TLSv1.3 ssl.provider = null ssl.secure.random.implementation = null ssl.trustmanager.algorithm = PKIX ssl.truststore.certificates = null ssl.truststore.location = null ssl.truststore.password = null ssl.truststore.type = JKS value.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer INFO 10:22:27.396[main] AssignorConfiguration - stream-thread [forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-consumer] Cooperative rebalancing enabled now INFO 10:22:27.411[main] AppInfoParser - Kafka version: 2.8.0 INFO 10:22:27.411[main] AppInfoParser - Kafka commitId: ebb1d6e21cc92130 INFO 10:22:27.411[main] AppInfoParser - Kafka startTimeMs: 1622535747411 INFO 10:22:27.422[main] ProducerConfig - ProducerConfig values: acks = 1 batch.size = 16384 bootstrap.servers = [localhost:9092] buffer.memory = 33554432 client.dns.lookup = use_all_dns_ips client.id = producer-1 compression.type = none connections.max.idle.ms = 540000 delivery.timeout.ms = 120000 enable.idempotence = false interceptor.classes = [] internal.auto.downgrade.txn.commit = false key.serializer = class org.apache.kafka.common.serialization.StringSerializer linger.ms = 0 max.block.ms = 60000 max.in.flight.requests.per.connection = 5 max.request.size = 1048576 metadata.max.age.ms = 300000 metadata.max.idle.ms = 300000 metric.reporters = [] metrics.num.samples = 2 metrics.recording.level = INFO metrics.sample.window.ms = 30000 partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner receive.buffer.bytes = 32768 reconnect.backoff.max.ms = 1000 reconnect.backoff.ms = 50 request.timeout.ms = 30000 retries = 2147483647 retry.backoff.ms = 100 sasl.client.callback.handler.class = null sasl.jaas.config = null sasl.kerberos.kinit.cmd = /usr/bin/kinit sasl.kerberos.min.time.before.relogin = 60000 sasl.kerberos.service.name = null sasl.kerberos.ticket.renew.jitter = 0.05 sasl.kerberos.ticket.renew.window.factor = 0.8 sasl.login.callback.handler.class = null sasl.login.class = null sasl.login.refresh.buffer.seconds = 300 sasl.login.refresh.min.period.seconds = 60 sasl.login.refresh.window.factor = 0.8 sasl.login.refresh.window.jitter = 0.05 sasl.mechanism = GSSAPI security.protocol = PLAINTEXT security.providers = null send.buffer.bytes = 131072 socket.connection.setup.timeout.max.ms = 30000 socket.connection.setup.timeout.ms = 10000 ssl.cipher.suites = null ssl.enabled.protocols = [TLSv1.2, TLSv1.3] ssl.endpoint.identification.algorithm = https ssl.engine.factory.class = null ssl.key.password = null ssl.keymanager.algorithm = SunX509 ssl.keystore.certificate.chain = null ssl.keystore.key = null ssl.keystore.location = null ssl.keystore.password = null ssl.keystore.type = JKS ssl.protocol = TLSv1.3 ssl.provider = null ssl.secure.random.implementation = null ssl.trustmanager.algorithm = PKIX ssl.truststore.certificates = null ssl.truststore.location = null ssl.truststore.password = null ssl.truststore.type = JKS transaction.timeout.ms = 60000 transactional.id = null value.serializer = class org.apache.kafka.common.serialization.StringSerializer INFO 10:22:27.438[main] AppInfoParser - Kafka version: 2.8.0 INFO 10:22:27.438[main] AppInfoParser - Kafka commitId: ebb1d6e21cc92130 INFO 10:22:27.438[main] AppInfoParser - Kafka startTimeMs: 1622535747438 DEBUG 10:22:27.438[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Starting Kafka producer I/O thread. INFO 10:22:27.439[main] Main - Starting... INFO 10:22:27.439[main] KafkaStreams - stream-client [forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c] State transition from CREATED to REBALANCING INFO 10:22:27.440[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] StreamThread - stream-thread [forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] Starting INFO 10:22:27.441[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] StreamThread - stream-thread [forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] State transition from CREATED to STARTING INFO 10:22:27.442[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] KafkaConsumer - [Consumer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-consumer, groupId=forwarding-stream] Subscribed to topic(s): input-topic INFO 10:22:27.590[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] StreamThread - stream-thread [forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] Processed 0 total records, ran 0 punctuators, and committed 0 total tasks since the last update INFO 10:22:27.596[kafka-producer-network-thread | producer-1] Metadata - [Producer clientId=producer-1] Cluster ID: VQfOfYG1TH6lbpjGN-sk5Q INFO 10:22:27.596[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] Metadata - [Consumer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-consumer, groupId=forwarding-stream] Cluster ID: VQfOfYG1TH6lbpjGN-sk5Q INFO 10:22:27.597[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] AbstractCoordinator - [Consumer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-consumer, groupId=forwarding-stream] Discovered group coordinator datorn:9092 (id: 2147483647 rack: null) INFO 10:22:27.599[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] AbstractCoordinator - [Consumer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-consumer, groupId=forwarding-stream] (Re-)joining group TRACE 10:22:27.602[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 59848ms INFO 10:22:27.607[main] TestProducer - Sent record WARN 10:22:27.608[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false WARN 10:22:27.610[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false WARN 10:22:27.610[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false WARN 10:22:27.611[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false INFO 10:22:27.613[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] AbstractCoordinator - [Consumer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-consumer, groupId=forwarding-stream] (Re-)joining group INFO 10:22:27.616[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] AbstractCoordinator - [Consumer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-consumer, groupId=forwarding-stream] Successfully joined group with generation Generation{generationId=5, memberId='forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-consumer-43fb25e4-449c-408d-bd93-6e56f83ad410', protocol='stream'} TRACE 10:22:27.617[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:27.620[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:22:27.622[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 3 INFO 10:22:27.642[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] HighAvailabilityTaskAssignor - Decided on assignment: {d0aed0d0-6f35-46a5-a601-07f4d727339c=[activeTasks: ([0_0]) standbyTasks: ([]) prevActiveTasks: ([]) prevStandbyTasks: ([]) changelogOffsetTotalsByTask: ([]) taskLagTotals: ([]) capacity: 1 assigned: 1]} with no followup probing rebalance. INFO 10:22:27.643[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] StreamsPartitionAssignor - stream-thread [forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-consumer] Assigned tasks [0_0] including stateful [] to clients as: d0aed0d0-6f35-46a5-a601-07f4d727339c=[activeTasks: ([0_0]) standbyTasks: ([])]. INFO 10:22:27.646[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] StreamsPartitionAssignor - stream-thread [forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-consumer] Client d0aed0d0-6f35-46a5-a601-07f4d727339c per-consumer assignment: prev owned active {} prev owned standby {forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-consumer-43fb25e4-449c-408d-bd93-6e56f83ad410=[]} assigned active {forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-consumer-43fb25e4-449c-408d-bd93-6e56f83ad410=[0_0]} revoking active {} assigned standby {} INFO 10:22:27.646[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] StreamsPartitionAssignor - stream-thread [forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-consumer] Finished stable assignment of tasks, no followup rebalances required. INFO 10:22:27.647[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] ConsumerCoordinator - [Consumer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-consumer, groupId=forwarding-stream] Finished assignment for group at generation 5: {forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-consumer-43fb25e4-449c-408d-bd93-6e56f83ad410=Assignment(partitions=[input-topic-0], userDataSize=48)} INFO 10:22:27.651[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] AbstractCoordinator - [Consumer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-consumer, groupId=forwarding-stream] Successfully synced group in generation Generation{generationId=5, memberId='forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-consumer-43fb25e4-449c-408d-bd93-6e56f83ad410', protocol='stream'} INFO 10:22:27.652[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] ConsumerCoordinator - [Consumer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-consumer, groupId=forwarding-stream] Updating assignment with Assigned partitions: [input-topic-0] Current owned partitions: [] Added partitions (assigned - owned): [input-topic-0] Revoked partitions (owned - assigned): [] INFO 10:22:27.652[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] ConsumerCoordinator - [Consumer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-consumer, groupId=forwarding-stream] Notifying assignor about the new Assignment(partitions=[input-topic-0], userDataSize=48) INFO 10:22:27.653[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] StreamsPartitionAssignor - stream-thread [forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-consumer] No followup rebalance was requested, resetting the rebalance schedule. INFO 10:22:27.654[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] TaskManager - stream-thread [forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] Handle new assignment with: New active tasks: [0_0] New standby tasks: [] Existing active tasks: [] Existing standby tasks: [] INFO 10:22:27.659[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] StreamThread - stream-thread [forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] Creating producer client for task 0_0 INFO 10:22:27.661[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] ProducerConfig - ProducerConfig values: acks = -1 batch.size = 102400 bootstrap.servers = [localhost:9092] buffer.memory = 33554432 client.dns.lookup = use_all_dns_ips client.id = forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer compression.type = none connections.max.idle.ms = 540000 delivery.timeout.ms = 2147483647 enable.idempotence = true interceptor.classes = [] internal.auto.downgrade.txn.commit = true key.serializer = class org.apache.kafka.common.serialization.ByteArraySerializer linger.ms = 5000 max.block.ms = 60000 max.in.flight.requests.per.connection = 5 max.request.size = 1048576 metadata.max.age.ms = 300000 metadata.max.idle.ms = 300000 metric.reporters = [] metrics.num.samples = 2 metrics.recording.level = INFO metrics.sample.window.ms = 30000 partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner receive.buffer.bytes = 32768 reconnect.backoff.max.ms = 1000 reconnect.backoff.ms = 50 request.timeout.ms = 30000 retries = 2147483647 retry.backoff.ms = 100 sasl.client.callback.handler.class = null sasl.jaas.config = null sasl.kerberos.kinit.cmd = /usr/bin/kinit sasl.kerberos.min.time.before.relogin = 60000 sasl.kerberos.service.name = null sasl.kerberos.ticket.renew.jitter = 0.05 sasl.kerberos.ticket.renew.window.factor = 0.8 sasl.login.callback.handler.class = null sasl.login.class = null sasl.login.refresh.buffer.seconds = 300 sasl.login.refresh.min.period.seconds = 60 sasl.login.refresh.window.factor = 0.8 sasl.login.refresh.window.jitter = 0.05 sasl.mechanism = GSSAPI security.protocol = PLAINTEXT security.providers = null send.buffer.bytes = 131072 socket.connection.setup.timeout.max.ms = 30000 socket.connection.setup.timeout.ms = 10000 ssl.cipher.suites = null ssl.enabled.protocols = [TLSv1.2, TLSv1.3] ssl.endpoint.identification.algorithm = https ssl.engine.factory.class = null ssl.key.password = null ssl.keymanager.algorithm = SunX509 ssl.keystore.certificate.chain = null ssl.keystore.key = null ssl.keystore.location = null ssl.keystore.password = null ssl.keystore.type = JKS ssl.protocol = TLSv1.3 ssl.provider = null ssl.secure.random.implementation = null ssl.trustmanager.algorithm = PKIX ssl.truststore.certificates = null ssl.truststore.location = null ssl.truststore.password = null ssl.truststore.type = JKS transaction.timeout.ms = 10000 transactional.id = forwarding-stream-0_0 value.serializer = class org.apache.kafka.common.serialization.ByteArraySerializer INFO 10:22:27.666[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] KafkaProducer - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Instantiated a transactional producer. INFO 10:22:27.670[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] KafkaProducer - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Overriding the default acks to all since idempotence is enabled. INFO 10:22:27.671[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] AppInfoParser - Kafka version: 2.8.0 DEBUG 10:22:27.671[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Starting Kafka producer I/O thread. INFO 10:22:27.672[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] AppInfoParser - Kafka commitId: ebb1d6e21cc92130 INFO 10:22:27.672[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] AppInfoParser - Kafka startTimeMs: 1622535747671 INFO 10:22:27.678[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Metadata - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Cluster ID: VQfOfYG1TH6lbpjGN-sk5Q INFO 10:22:27.685[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] ConsumerCoordinator - [Consumer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-consumer, groupId=forwarding-stream] Adding newly assigned partitions: input-topic-0 INFO 10:22:27.685[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] StreamThread - stream-thread [forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] State transition from STARTING to PARTITIONS_ASSIGNED INFO 10:22:27.694[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] TransactionManager - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Invoking InitProducerId for the first time in order to acquire a producer ID TRACE 10:22:27.694[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Coordinator not known for TRANSACTION, will retry INIT_PRODUCER_ID after finding coordinator. DEBUG 10:22:27.698[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sending transactional request FindCoordinatorRequestData(key='forwarding-stream-0_0', keyType=1) to node datorn:9092 (id: 0 rack: null) with correlation ID 3 INFO 10:22:27.699[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] TransactionManager - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Discovered transaction coordinator datorn:9092 (id: 0 rack: null) DEBUG 10:22:27.800[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sending transactional request InitProducerIdRequestData(transactionalId='forwarding-stream-0_0', transactionTimeoutMs=10000, producerId=-1, producerEpoch=-1) to node datorn:9092 (id: 0 rack: null) with correlation ID 4 INFO 10:22:27.803[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] TransactionManager - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] ProducerId set to 0 with epoch 3 INFO 10:22:27.804[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] StreamTask - stream-thread [forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] task [0_0] Initialized INFO 10:22:27.812[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] StreamTask - stream-thread [forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] task [0_0] Restored and ready to run INFO 10:22:27.813[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] StreamThread - stream-thread [forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] Restoration took 127 ms for all tasks [0_0] INFO 10:22:27.813[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] StreamThread - stream-thread [forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] State transition from PARTITIONS_ASSIGNED to RUNNING INFO 10:22:27.814[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] KafkaStreams - stream-client [forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c] State transition from REBALANCING to RUNNING INFO 10:22:27.819[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] ConsumerCoordinator - [Consumer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-consumer, groupId=forwarding-stream] Setting offset for partition input-topic-0 to the committed offset FetchPosition{offset=154, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=Optional[datorn:9092 (id: 0 rack: null)], epoch=0}} TRACE 10:22:27.843[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 59995ms DEBUG 10:22:27.846[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sending transactional request AddPartitionsToTxnRequestData(transactionalId='forwarding-stream-0_0', producerId=0, producerEpoch=3, topics=[AddPartitionsToTxnTopic(name='output-topic', partitions=[0])]) to node datorn:9092 (id: 0 rack: null) with correlation ID 6 TRACE 10:22:28.608[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:22:28.608[main] TestProducer - Sent record WARN 10:22:28.608[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:22:28.609[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:28.609[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:22:28.611[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 4 TRACE 10:22:29.609[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:22:29.609[main] TestProducer - Sent record WARN 10:22:29.609[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:22:29.610[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:29.610[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:22:29.612[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 5 TRACE 10:22:30.609[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:22:30.610[main] TestProducer - Sent record WARN 10:22:30.610[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:22:30.610[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:30.611[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:22:30.613[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 6 TRACE 10:22:31.610[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:22:31.611[main] TestProducer - Sent record WARN 10:22:31.611[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:22:31.611[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:31.612[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:22:31.614[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 7 TRACE 10:22:32.611[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:22:32.611[main] TestProducer - Sent record WARN 10:22:32.612[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:22:32.612[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:32.612[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:22:32.614[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 8 WARN 10:22:32.848[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false DEBUG 10:22:32.851[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 0 being sent to partition output-topic-0 TRACE 10:22:32.852[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:32.853[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=156, buffer=java.nio.HeapByteBuffer[pos=0 lim=156 cap=156]))]), transactionalId='forwarding-stream-0_0' TRACE 10:22:32.855[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 7 TRACE 10:22:33.612[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:22:33.612[main] TestProducer - Sent record WARN 10:22:33.612[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:22:33.613[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:33.613[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:22:33.615[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 9 TRACE 10:22:33.618[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms TRACE 10:22:34.612[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:22:34.613[main] TestProducer - Sent record WARN 10:22:34.613[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:22:34.613[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:34.614[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:22:34.616[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 10 TRACE 10:22:35.613[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:22:35.614[main] TestProducer - Sent record WARN 10:22:35.614[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:22:35.614[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:35.615[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:22:35.617[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 11 TRACE 10:22:36.614[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:22:36.614[main] TestProducer - Sent record WARN 10:22:36.614[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:22:36.615[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:36.615[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:22:36.617[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 12 TRACE 10:22:37.615[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:22:37.615[main] TestProducer - Sent record WARN 10:22:37.615[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:22:37.616[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:37.616[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:22:37.618[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 13 WARN 10:22:37.822[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] beginFlush(): 0 -> 1 java.lang.Exception: null at org.apache.kafka.clients.producer.internals.RecordAccumulator.beginFlush(RecordAccumulator.java:701) at org.apache.kafka.clients.producer.KafkaProducer.flush(KafkaProducer.java:1123) at org.apache.kafka.streams.processor.internals.StreamsProducer.flush(StreamsProducer.java:316) at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.flush(RecordCollectorImpl.java:258) at org.apache.kafka.streams.processor.internals.StreamTask.prepareCommit(StreamTask.java:391) at org.apache.kafka.streams.processor.internals.TaskManager.commitAndFillInConsumedOffsetsAndMetadataPerTaskMap(TaskManager.java:1052) at org.apache.kafka.streams.processor.internals.TaskManager.commit(TaskManager.java:1025) at org.apache.kafka.streams.processor.internals.StreamThread.maybeCommit(StreamThread.java:1010) at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:786) at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:583) at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:556) WARN 10:22:37.827[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:22:37.827[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 6 being sent to partition output-topic-0 TRACE 10:22:37.827[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:37.827[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=140, buffer=java.nio.HeapByteBuffer[pos=0 lim=140 cap=140]))]), transactionalId='forwarding-stream-0_0' TRACE 10:22:37.830[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 8 WARN 10:22:37.830[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] awaitFlushCompletion(): 1 -> 0 java.lang.Exception: null at org.apache.kafka.clients.producer.internals.RecordAccumulator.awaitFlushCompletion(RecordAccumulator.java:720) at org.apache.kafka.clients.producer.KafkaProducer.flush(KafkaProducer.java:1126) at org.apache.kafka.streams.processor.internals.StreamsProducer.flush(StreamsProducer.java:316) at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.flush(RecordCollectorImpl.java:258) at org.apache.kafka.streams.processor.internals.StreamTask.prepareCommit(StreamTask.java:391) at org.apache.kafka.streams.processor.internals.TaskManager.commitAndFillInConsumedOffsetsAndMetadataPerTaskMap(TaskManager.java:1052) at org.apache.kafka.streams.processor.internals.TaskManager.commit(TaskManager.java:1025) at org.apache.kafka.streams.processor.internals.StreamThread.maybeCommit(StreamThread.java:1010) at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:786) at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:583) at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:556) DEBUG 10:22:37.831[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sending transactional request AddOffsetsToTxnRequestData(transactionalId='forwarding-stream-0_0', producerId=0, producerEpoch=3, groupId='forwarding-stream') to node datorn:9092 (id: 0 rack: null) with correlation ID 9 TRACE 10:22:37.836[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Coordinator not known for GROUP, will retry TXN_OFFSET_COMMIT after finding coordinator. DEBUG 10:22:37.837[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sending transactional request FindCoordinatorRequestData(key='forwarding-stream', keyType=0) to node datorn:9092 (id: 0 rack: null) with correlation ID 10 INFO 10:22:37.838[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] TransactionManager - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Discovered group coordinator datorn:9092 (id: 0 rack: null) DEBUG 10:22:37.938[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sending transactional request TxnOffsetCommitRequestData(transactionalId='forwarding-stream-0_0', groupId='forwarding-stream', producerId=0, producerEpoch=3, generationId=5, memberId='forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-consumer-43fb25e4-449c-408d-bd93-6e56f83ad410', groupInstanceId=null, topics=[TxnOffsetCommitRequestTopic(name='input-topic', partitions=[TxnOffsetCommitRequestPartition(partitionIndex=0, committedOffset=165, committedLeaderEpoch=-1, committedMetadata='AQAAAXnGqt8u')])]) to node datorn:9092 (id: 0 rack: null) with correlation ID 11 WARN 10:22:37.943[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] beginFlush(): 0 -> 1 java.lang.Exception: null at org.apache.kafka.clients.producer.internals.RecordAccumulator.beginFlush(RecordAccumulator.java:701) at org.apache.kafka.clients.producer.internals.Sender.maybeSendAndPollTransactionalRequest(Sender.java:439) at org.apache.kafka.clients.producer.internals.Sender.runOnce(Sender.java:315) at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:242) at java.base/java.lang.Thread.run(Thread.java:829) DEBUG 10:22:37.944[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sending transactional request EndTxnRequestData(transactionalId='forwarding-stream-0_0', producerId=0, producerEpoch=3, committed=true) to node datorn:9092 (id: 0 rack: null) with correlation ID 12 TRACE 10:22:38.616[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:22:38.616[main] TestProducer - Sent record WARN 10:22:38.616[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:22:38.616[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:38.617[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:22:38.619[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 14 TRACE 10:22:38.621[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms DEBUG 10:22:38.622[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sending transactional request AddPartitionsToTxnRequestData(transactionalId='forwarding-stream-0_0', producerId=0, producerEpoch=3, topics=[AddPartitionsToTxnTopic(name='output-topic', partitions=[0])]) to node datorn:9092 (id: 0 rack: null) with correlation ID 13 WARN 10:22:38.624[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:22:38.624[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 11 being sent to partition output-topic-0 TRACE 10:22:38.625[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:38.625[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:22:38.627[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 14 TRACE 10:22:39.616[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:22:39.616[main] TestProducer - Sent record WARN 10:22:39.617[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:22:39.617[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:39.618[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:22:39.619[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 15 TRACE 10:22:39.622[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:22:39.622[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:22:39.622[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 12 being sent to partition output-topic-0 TRACE 10:22:39.623[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:39.623[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:22:39.625[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 15 TRACE 10:22:40.617[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:22:40.617[main] TestProducer - Sent record WARN 10:22:40.617[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:22:40.617[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:40.618[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:22:40.620[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 16 TRACE 10:22:40.622[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:22:40.622[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:22:40.623[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 13 being sent to partition output-topic-0 TRACE 10:22:40.623[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:40.623[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:22:40.625[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 16 TRACE 10:22:41.617[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:22:41.618[main] TestProducer - Sent record WARN 10:22:41.618[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:22:41.618[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:41.619[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:22:41.620[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 17 TRACE 10:22:41.622[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:22:41.623[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:22:41.623[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 14 being sent to partition output-topic-0 TRACE 10:22:41.623[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:41.623[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:22:41.625[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 17 TRACE 10:22:42.618[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:22:42.618[main] TestProducer - Sent record WARN 10:22:42.618[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:22:42.619[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:42.619[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:22:42.621[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 18 TRACE 10:22:42.623[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:22:42.624[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:22:42.624[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 15 being sent to partition output-topic-0 TRACE 10:22:42.624[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:42.625[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:22:42.626[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 18 TRACE 10:22:43.619[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:22:43.619[main] TestProducer - Sent record WARN 10:22:43.619[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:22:43.619[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:43.620[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:22:43.622[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 19 TRACE 10:22:43.623[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:22:43.624[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:22:43.624[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 16 being sent to partition output-topic-0 TRACE 10:22:43.624[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:43.625[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:22:43.626[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 19 TRACE 10:22:44.619[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:22:44.619[main] TestProducer - Sent record WARN 10:22:44.620[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:22:44.620[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:44.620[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:22:44.622[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 20 TRACE 10:22:44.624[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:22:44.625[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:22:44.625[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 17 being sent to partition output-topic-0 TRACE 10:22:44.625[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:44.626[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:22:44.628[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 20 TRACE 10:22:45.620[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:22:45.620[main] TestProducer - Sent record WARN 10:22:45.620[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:22:45.621[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:45.621[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:22:45.623[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 21 TRACE 10:22:45.625[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:22:45.626[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:22:45.626[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 18 being sent to partition output-topic-0 TRACE 10:22:45.626[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:45.627[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:22:45.628[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 21 TRACE 10:22:46.620[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:22:46.621[main] TestProducer - Sent record WARN 10:22:46.621[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:22:46.621[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:46.622[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:22:46.624[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 22 TRACE 10:22:46.627[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:22:46.627[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:22:46.627[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 19 being sent to partition output-topic-0 TRACE 10:22:46.628[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:46.628[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:22:46.630[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 22 TRACE 10:22:47.621[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:22:47.622[main] TestProducer - Sent record WARN 10:22:47.622[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:22:47.622[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:47.623[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:22:47.624[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 23 TRACE 10:22:47.627[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:22:47.627[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:22:47.627[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 20 being sent to partition output-topic-0 TRACE 10:22:47.627[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:47.628[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:22:47.630[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 23 WARN 10:22:48.029[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] beginFlush(): 1 -> 2 java.lang.Exception: null at org.apache.kafka.clients.producer.internals.RecordAccumulator.beginFlush(RecordAccumulator.java:701) at org.apache.kafka.clients.producer.KafkaProducer.flush(KafkaProducer.java:1123) at org.apache.kafka.streams.processor.internals.StreamsProducer.flush(StreamsProducer.java:316) at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.flush(RecordCollectorImpl.java:258) at org.apache.kafka.streams.processor.internals.StreamTask.prepareCommit(StreamTask.java:391) at org.apache.kafka.streams.processor.internals.TaskManager.commitAndFillInConsumedOffsetsAndMetadataPerTaskMap(TaskManager.java:1052) at org.apache.kafka.streams.processor.internals.TaskManager.commit(TaskManager.java:1025) at org.apache.kafka.streams.processor.internals.StreamThread.maybeCommit(StreamThread.java:1010) at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:786) at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:583) at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:556) WARN 10:22:48.030[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] awaitFlushCompletion(): 2 -> 1 java.lang.Exception: null at org.apache.kafka.clients.producer.internals.RecordAccumulator.awaitFlushCompletion(RecordAccumulator.java:720) at org.apache.kafka.clients.producer.KafkaProducer.flush(KafkaProducer.java:1126) at org.apache.kafka.streams.processor.internals.StreamsProducer.flush(StreamsProducer.java:316) at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.flush(RecordCollectorImpl.java:258) at org.apache.kafka.streams.processor.internals.StreamTask.prepareCommit(StreamTask.java:391) at org.apache.kafka.streams.processor.internals.TaskManager.commitAndFillInConsumedOffsetsAndMetadataPerTaskMap(TaskManager.java:1052) at org.apache.kafka.streams.processor.internals.TaskManager.commit(TaskManager.java:1025) at org.apache.kafka.streams.processor.internals.StreamThread.maybeCommit(StreamThread.java:1010) at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:786) at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:583) at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:556) DEBUG 10:22:48.030[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sending transactional request AddOffsetsToTxnRequestData(transactionalId='forwarding-stream-0_0', producerId=0, producerEpoch=3, groupId='forwarding-stream') to node datorn:9092 (id: 0 rack: null) with correlation ID 24 DEBUG 10:22:48.032[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sending transactional request TxnOffsetCommitRequestData(transactionalId='forwarding-stream-0_0', groupId='forwarding-stream', producerId=0, producerEpoch=3, generationId=5, memberId='forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-consumer-43fb25e4-449c-408d-bd93-6e56f83ad410', groupInstanceId=null, topics=[TxnOffsetCommitRequestTopic(name='input-topic', partitions=[TxnOffsetCommitRequestPartition(partitionIndex=0, committedOffset=175, committedLeaderEpoch=-1, committedMetadata='AQAAAXnGqwZF')])]) to node datorn:9092 (id: 0 rack: null) with correlation ID 25 DEBUG 10:22:48.034[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sending transactional request EndTxnRequestData(transactionalId='forwarding-stream-0_0', producerId=0, producerEpoch=3, committed=true) to node datorn:9092 (id: 0 rack: null) with correlation ID 26 TRACE 10:22:48.622[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:22:48.622[main] TestProducer - Sent record WARN 10:22:48.623[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:22:48.623[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:48.623[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:22:48.625[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 24 TRACE 10:22:48.627[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms DEBUG 10:22:48.628[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sending transactional request AddPartitionsToTxnRequestData(transactionalId='forwarding-stream-0_0', producerId=0, producerEpoch=3, topics=[AddPartitionsToTxnTopic(name='output-topic', partitions=[0])]) to node datorn:9092 (id: 0 rack: null) with correlation ID 27 WARN 10:22:48.630[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:22:48.630[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 21 being sent to partition output-topic-0 TRACE 10:22:48.631[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:48.631[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:22:48.632[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 28 TRACE 10:22:49.623[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:22:49.623[main] TestProducer - Sent record WARN 10:22:49.623[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:22:49.623[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:49.624[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:22:49.626[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 25 TRACE 10:22:49.628[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:22:49.628[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:22:49.628[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 22 being sent to partition output-topic-0 TRACE 10:22:49.629[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:49.629[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:22:49.631[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 29 TRACE 10:22:50.623[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:22:50.624[main] TestProducer - Sent record WARN 10:22:50.624[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:22:50.624[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:50.624[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:22:50.626[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 26 TRACE 10:22:50.628[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:22:50.629[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:22:50.629[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 23 being sent to partition output-topic-0 TRACE 10:22:50.629[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:50.630[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:22:50.631[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 30 TRACE 10:22:51.624[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:22:51.624[main] TestProducer - Sent record WARN 10:22:51.624[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:22:51.625[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:51.625[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:22:51.627[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 27 TRACE 10:22:51.629[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:22:51.629[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:22:51.629[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 24 being sent to partition output-topic-0 TRACE 10:22:51.629[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:51.630[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:22:51.631[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 31 TRACE 10:22:52.625[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:22:52.625[main] TestProducer - Sent record WARN 10:22:52.625[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:22:52.625[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:52.626[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:22:52.627[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 28 TRACE 10:22:52.629[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:22:52.630[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:22:52.630[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 25 being sent to partition output-topic-0 TRACE 10:22:52.630[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:52.631[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:22:52.632[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 32 TRACE 10:22:53.625[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:22:53.625[main] TestProducer - Sent record WARN 10:22:53.626[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:22:53.626[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:53.626[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:22:53.628[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 29 TRACE 10:22:53.630[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:22:53.630[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:22:53.630[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 26 being sent to partition output-topic-0 TRACE 10:22:53.631[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:53.631[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:22:53.633[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 33 TRACE 10:22:54.626[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:22:54.626[main] TestProducer - Sent record WARN 10:22:54.626[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:22:54.626[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:54.627[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:22:54.629[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 30 TRACE 10:22:54.631[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:22:54.631[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:22:54.631[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 27 being sent to partition output-topic-0 TRACE 10:22:54.631[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:54.632[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:22:54.633[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 34 TRACE 10:22:55.626[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:22:55.627[main] TestProducer - Sent record WARN 10:22:55.627[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:22:55.627[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:55.627[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:22:55.629[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 31 TRACE 10:22:55.631[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:22:55.632[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:22:55.632[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 28 being sent to partition output-topic-0 TRACE 10:22:55.632[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:55.632[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:22:55.634[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 35 TRACE 10:22:56.627[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:22:56.627[main] TestProducer - Sent record WARN 10:22:56.627[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:22:56.628[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:56.628[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:22:56.630[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 32 TRACE 10:22:56.632[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:22:56.633[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:22:56.633[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 29 being sent to partition output-topic-0 TRACE 10:22:56.633[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:56.634[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:22:56.635[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 36 TRACE 10:22:57.628[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:22:57.628[main] TestProducer - Sent record WARN 10:22:57.628[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:22:57.628[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:57.629[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:22:57.631[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 33 TRACE 10:22:57.632[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:22:57.633[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:22:57.633[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 30 being sent to partition output-topic-0 TRACE 10:22:57.633[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:57.634[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:22:57.635[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 37 WARN 10:22:58.134[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] beginFlush(): 1 -> 2 java.lang.Exception: null at org.apache.kafka.clients.producer.internals.RecordAccumulator.beginFlush(RecordAccumulator.java:701) at org.apache.kafka.clients.producer.KafkaProducer.flush(KafkaProducer.java:1123) at org.apache.kafka.streams.processor.internals.StreamsProducer.flush(StreamsProducer.java:316) at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.flush(RecordCollectorImpl.java:258) at org.apache.kafka.streams.processor.internals.StreamTask.prepareCommit(StreamTask.java:391) at org.apache.kafka.streams.processor.internals.TaskManager.commitAndFillInConsumedOffsetsAndMetadataPerTaskMap(TaskManager.java:1052) at org.apache.kafka.streams.processor.internals.TaskManager.commit(TaskManager.java:1025) at org.apache.kafka.streams.processor.internals.StreamThread.maybeCommit(StreamThread.java:1010) at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:786) at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:583) at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:556) WARN 10:22:58.134[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] awaitFlushCompletion(): 2 -> 1 java.lang.Exception: null at org.apache.kafka.clients.producer.internals.RecordAccumulator.awaitFlushCompletion(RecordAccumulator.java:720) at org.apache.kafka.clients.producer.KafkaProducer.flush(KafkaProducer.java:1126) at org.apache.kafka.streams.processor.internals.StreamsProducer.flush(StreamsProducer.java:316) at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.flush(RecordCollectorImpl.java:258) at org.apache.kafka.streams.processor.internals.StreamTask.prepareCommit(StreamTask.java:391) at org.apache.kafka.streams.processor.internals.TaskManager.commitAndFillInConsumedOffsetsAndMetadataPerTaskMap(TaskManager.java:1052) at org.apache.kafka.streams.processor.internals.TaskManager.commit(TaskManager.java:1025) at org.apache.kafka.streams.processor.internals.StreamThread.maybeCommit(StreamThread.java:1010) at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:786) at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:583) at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:556) DEBUG 10:22:58.135[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sending transactional request AddOffsetsToTxnRequestData(transactionalId='forwarding-stream-0_0', producerId=0, producerEpoch=3, groupId='forwarding-stream') to node datorn:9092 (id: 0 rack: null) with correlation ID 38 DEBUG 10:22:58.137[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sending transactional request TxnOffsetCommitRequestData(transactionalId='forwarding-stream-0_0', groupId='forwarding-stream', producerId=0, producerEpoch=3, generationId=5, memberId='forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-consumer-43fb25e4-449c-408d-bd93-6e56f83ad410', groupInstanceId=null, topics=[TxnOffsetCommitRequestTopic(name='input-topic', partitions=[TxnOffsetCommitRequestPartition(partitionIndex=0, committedOffset=185, committedLeaderEpoch=-1, committedMetadata='AQAAAXnGqy1b')])]) to node datorn:9092 (id: 0 rack: null) with correlation ID 39 DEBUG 10:22:58.140[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sending transactional request EndTxnRequestData(transactionalId='forwarding-stream-0_0', producerId=0, producerEpoch=3, committed=true) to node datorn:9092 (id: 0 rack: null) with correlation ID 40 TRACE 10:22:58.628[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:22:58.628[main] TestProducer - Sent record WARN 10:22:58.629[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:22:58.629[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:58.629[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:22:58.631[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 34 TRACE 10:22:58.633[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms DEBUG 10:22:58.634[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sending transactional request AddPartitionsToTxnRequestData(transactionalId='forwarding-stream-0_0', producerId=0, producerEpoch=3, topics=[AddPartitionsToTxnTopic(name='output-topic', partitions=[0])]) to node datorn:9092 (id: 0 rack: null) with correlation ID 41 WARN 10:22:58.636[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:22:58.636[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 31 being sent to partition output-topic-0 TRACE 10:22:58.636[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:58.637[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:22:58.638[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 42 TRACE 10:22:59.629[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:22:59.629[main] TestProducer - Sent record WARN 10:22:59.629[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:22:59.629[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:59.630[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:22:59.632[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 35 TRACE 10:22:59.633[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:22:59.634[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:22:59.634[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 32 being sent to partition output-topic-0 TRACE 10:22:59.634[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:22:59.635[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:22:59.637[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 43 TRACE 10:23:00.629[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:23:00.629[main] TestProducer - Sent record WARN 10:23:00.630[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:23:00.630[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:00.630[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:23:00.632[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 36 TRACE 10:23:00.634[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:23:00.634[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:23:00.634[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 33 being sent to partition output-topic-0 TRACE 10:23:00.635[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:00.635[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:23:00.637[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 44 TRACE 10:23:01.630[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:23:01.630[main] TestProducer - Sent record WARN 10:23:01.630[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:23:01.631[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:01.632[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:23:01.633[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 37 TRACE 10:23:01.635[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:23:01.636[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:23:01.636[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 34 being sent to partition output-topic-0 TRACE 10:23:01.636[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:01.636[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:23:01.638[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 45 TRACE 10:23:02.630[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:23:02.631[main] TestProducer - Sent record WARN 10:23:02.631[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:23:02.631[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:02.631[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:23:02.633[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 38 TRACE 10:23:02.636[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:23:02.636[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:23:02.636[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 35 being sent to partition output-topic-0 TRACE 10:23:02.636[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:02.637[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:23:02.639[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 46 TRACE 10:23:03.631[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:23:03.631[main] TestProducer - Sent record WARN 10:23:03.631[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:23:03.632[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:03.632[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:23:03.634[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 39 TRACE 10:23:03.637[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:23:03.638[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:23:03.638[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 36 being sent to partition output-topic-0 TRACE 10:23:03.638[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:03.638[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:23:03.640[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 47 TRACE 10:23:04.632[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:23:04.632[main] TestProducer - Sent record WARN 10:23:04.632[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:23:04.632[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:04.633[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:23:04.634[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 40 TRACE 10:23:04.636[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:23:04.637[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:23:04.637[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 37 being sent to partition output-topic-0 TRACE 10:23:04.637[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:04.637[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:23:04.639[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 48 TRACE 10:23:05.632[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:23:05.632[main] TestProducer - Sent record WARN 10:23:05.633[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:23:05.633[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:05.633[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:23:05.635[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 41 TRACE 10:23:05.637[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:23:05.637[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:23:05.637[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 38 being sent to partition output-topic-0 TRACE 10:23:05.637[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:05.638[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:23:05.639[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 49 TRACE 10:23:06.633[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:23:06.633[main] TestProducer - Sent record WARN 10:23:06.633[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:23:06.633[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:06.634[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:23:06.636[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 42 TRACE 10:23:06.638[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:23:06.639[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:23:06.639[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 39 being sent to partition output-topic-0 TRACE 10:23:06.639[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:06.639[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:23:06.640[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 50 TRACE 10:23:07.633[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:23:07.633[main] TestProducer - Sent record WARN 10:23:07.634[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:23:07.634[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:07.634[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:23:07.636[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 43 TRACE 10:23:07.638[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:23:07.638[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:23:07.639[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 40 being sent to partition output-topic-0 TRACE 10:23:07.639[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:07.639[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:23:07.641[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 51 WARN 10:23:08.241[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] beginFlush(): 1 -> 2 java.lang.Exception: null at org.apache.kafka.clients.producer.internals.RecordAccumulator.beginFlush(RecordAccumulator.java:701) at org.apache.kafka.clients.producer.KafkaProducer.flush(KafkaProducer.java:1123) at org.apache.kafka.streams.processor.internals.StreamsProducer.flush(StreamsProducer.java:316) at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.flush(RecordCollectorImpl.java:258) at org.apache.kafka.streams.processor.internals.StreamTask.prepareCommit(StreamTask.java:391) at org.apache.kafka.streams.processor.internals.TaskManager.commitAndFillInConsumedOffsetsAndMetadataPerTaskMap(TaskManager.java:1052) at org.apache.kafka.streams.processor.internals.TaskManager.commit(TaskManager.java:1025) at org.apache.kafka.streams.processor.internals.StreamThread.maybeCommit(StreamThread.java:1010) at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:786) at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:583) at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:556) WARN 10:23:08.241[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] awaitFlushCompletion(): 2 -> 1 java.lang.Exception: null at org.apache.kafka.clients.producer.internals.RecordAccumulator.awaitFlushCompletion(RecordAccumulator.java:720) at org.apache.kafka.clients.producer.KafkaProducer.flush(KafkaProducer.java:1126) at org.apache.kafka.streams.processor.internals.StreamsProducer.flush(StreamsProducer.java:316) at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.flush(RecordCollectorImpl.java:258) at org.apache.kafka.streams.processor.internals.StreamTask.prepareCommit(StreamTask.java:391) at org.apache.kafka.streams.processor.internals.TaskManager.commitAndFillInConsumedOffsetsAndMetadataPerTaskMap(TaskManager.java:1052) at org.apache.kafka.streams.processor.internals.TaskManager.commit(TaskManager.java:1025) at org.apache.kafka.streams.processor.internals.StreamThread.maybeCommit(StreamThread.java:1010) at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:786) at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:583) at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:556) DEBUG 10:23:08.242[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sending transactional request AddOffsetsToTxnRequestData(transactionalId='forwarding-stream-0_0', producerId=0, producerEpoch=3, groupId='forwarding-stream') to node datorn:9092 (id: 0 rack: null) with correlation ID 52 DEBUG 10:23:08.244[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sending transactional request TxnOffsetCommitRequestData(transactionalId='forwarding-stream-0_0', groupId='forwarding-stream', producerId=0, producerEpoch=3, generationId=5, memberId='forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-consumer-43fb25e4-449c-408d-bd93-6e56f83ad410', groupInstanceId=null, topics=[TxnOffsetCommitRequestTopic(name='input-topic', partitions=[TxnOffsetCommitRequestPartition(partitionIndex=0, committedOffset=195, committedLeaderEpoch=-1, committedMetadata='AQAAAXnGq1Rx')])]) to node datorn:9092 (id: 0 rack: null) with correlation ID 53 DEBUG 10:23:08.246[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sending transactional request EndTxnRequestData(transactionalId='forwarding-stream-0_0', producerId=0, producerEpoch=3, committed=true) to node datorn:9092 (id: 0 rack: null) with correlation ID 54 TRACE 10:23:08.634[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:23:08.634[main] TestProducer - Sent record WARN 10:23:08.634[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:23:08.634[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:08.635[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:23:08.637[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 44 TRACE 10:23:08.639[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms DEBUG 10:23:08.639[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sending transactional request AddPartitionsToTxnRequestData(transactionalId='forwarding-stream-0_0', producerId=0, producerEpoch=3, topics=[AddPartitionsToTxnTopic(name='output-topic', partitions=[0])]) to node datorn:9092 (id: 0 rack: null) with correlation ID 55 WARN 10:23:08.642[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:23:08.642[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 41 being sent to partition output-topic-0 TRACE 10:23:08.642[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:08.643[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:23:08.644[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 56 TRACE 10:23:09.634[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:23:09.635[main] TestProducer - Sent record WARN 10:23:09.635[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:23:09.635[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:09.636[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:23:09.637[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 45 TRACE 10:23:09.639[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:23:09.640[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:23:09.640[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 42 being sent to partition output-topic-0 TRACE 10:23:09.640[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:09.640[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:23:09.642[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 57 TRACE 10:23:10.635[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:23:10.635[main] TestProducer - Sent record WARN 10:23:10.635[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:23:10.636[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:10.636[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:23:10.638[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 46 TRACE 10:23:10.640[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:23:10.641[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:23:10.641[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 43 being sent to partition output-topic-0 TRACE 10:23:10.641[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:10.642[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:23:10.643[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 58 TRACE 10:23:11.636[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:23:11.636[main] TestProducer - Sent record WARN 10:23:11.636[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:23:11.636[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:11.637[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:23:11.638[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 47 TRACE 10:23:11.641[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:23:11.641[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:23:11.641[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 44 being sent to partition output-topic-0 TRACE 10:23:11.641[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:11.642[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:23:11.643[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 59 TRACE 10:23:12.636[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:23:12.637[main] TestProducer - Sent record WARN 10:23:12.637[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:23:12.637[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:12.637[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:23:12.639[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 48 TRACE 10:23:12.641[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:23:12.641[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:23:12.641[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 45 being sent to partition output-topic-0 TRACE 10:23:12.641[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:12.642[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:23:12.643[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 60 TRACE 10:23:13.637[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:23:13.637[main] TestProducer - Sent record WARN 10:23:13.638[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:23:13.638[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:13.638[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:23:13.640[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 49 TRACE 10:23:13.641[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:23:13.642[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:23:13.642[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 46 being sent to partition output-topic-0 TRACE 10:23:13.642[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:13.643[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:23:13.644[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 61 TRACE 10:23:14.638[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:23:14.638[main] TestProducer - Sent record WARN 10:23:14.638[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:23:14.638[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:14.639[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:23:14.640[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 50 TRACE 10:23:14.642[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:23:14.642[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:23:14.642[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 47 being sent to partition output-topic-0 TRACE 10:23:14.642[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:14.643[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:23:14.644[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 62 TRACE 10:23:15.638[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:23:15.638[main] TestProducer - Sent record WARN 10:23:15.639[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:23:15.639[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:15.639[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:23:15.640[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 51 TRACE 10:23:15.643[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:23:15.643[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:23:15.644[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 48 being sent to partition output-topic-0 TRACE 10:23:15.644[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:15.644[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:23:15.646[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 63 TRACE 10:23:16.639[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:23:16.639[main] TestProducer - Sent record WARN 10:23:16.639[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:23:16.639[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:16.640[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:23:16.642[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 52 TRACE 10:23:16.643[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:23:16.644[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:23:16.644[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 49 being sent to partition output-topic-0 TRACE 10:23:16.644[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:16.644[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:23:16.646[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 64 TRACE 10:23:17.639[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:23:17.640[main] TestProducer - Sent record WARN 10:23:17.640[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:23:17.640[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:17.640[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:23:17.642[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 53 TRACE 10:23:17.644[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:23:17.645[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:23:17.645[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 50 being sent to partition output-topic-0 TRACE 10:23:17.645[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:17.645[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:23:17.647[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 65 WARN 10:23:18.348[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] beginFlush(): 1 -> 2 java.lang.Exception: null at org.apache.kafka.clients.producer.internals.RecordAccumulator.beginFlush(RecordAccumulator.java:701) at org.apache.kafka.clients.producer.KafkaProducer.flush(KafkaProducer.java:1123) at org.apache.kafka.streams.processor.internals.StreamsProducer.flush(StreamsProducer.java:316) at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.flush(RecordCollectorImpl.java:258) at org.apache.kafka.streams.processor.internals.StreamTask.prepareCommit(StreamTask.java:391) at org.apache.kafka.streams.processor.internals.TaskManager.commitAndFillInConsumedOffsetsAndMetadataPerTaskMap(TaskManager.java:1052) at org.apache.kafka.streams.processor.internals.TaskManager.commit(TaskManager.java:1025) at org.apache.kafka.streams.processor.internals.StreamThread.maybeCommit(StreamThread.java:1010) at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:786) at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:583) at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:556) WARN 10:23:18.348[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] awaitFlushCompletion(): 2 -> 1 java.lang.Exception: null at org.apache.kafka.clients.producer.internals.RecordAccumulator.awaitFlushCompletion(RecordAccumulator.java:720) at org.apache.kafka.clients.producer.KafkaProducer.flush(KafkaProducer.java:1126) at org.apache.kafka.streams.processor.internals.StreamsProducer.flush(StreamsProducer.java:316) at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.flush(RecordCollectorImpl.java:258) at org.apache.kafka.streams.processor.internals.StreamTask.prepareCommit(StreamTask.java:391) at org.apache.kafka.streams.processor.internals.TaskManager.commitAndFillInConsumedOffsetsAndMetadataPerTaskMap(TaskManager.java:1052) at org.apache.kafka.streams.processor.internals.TaskManager.commit(TaskManager.java:1025) at org.apache.kafka.streams.processor.internals.StreamThread.maybeCommit(StreamThread.java:1010) at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:786) at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:583) at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:556) DEBUG 10:23:18.349[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sending transactional request AddOffsetsToTxnRequestData(transactionalId='forwarding-stream-0_0', producerId=0, producerEpoch=3, groupId='forwarding-stream') to node datorn:9092 (id: 0 rack: null) with correlation ID 66 DEBUG 10:23:18.351[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sending transactional request TxnOffsetCommitRequestData(transactionalId='forwarding-stream-0_0', groupId='forwarding-stream', producerId=0, producerEpoch=3, generationId=5, memberId='forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-consumer-43fb25e4-449c-408d-bd93-6e56f83ad410', groupInstanceId=null, topics=[TxnOffsetCommitRequestTopic(name='input-topic', partitions=[TxnOffsetCommitRequestPartition(partitionIndex=0, committedOffset=205, committedLeaderEpoch=-1, committedMetadata='AQAAAXnGq3uH')])]) to node datorn:9092 (id: 0 rack: null) with correlation ID 67 DEBUG 10:23:18.353[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sending transactional request EndTxnRequestData(transactionalId='forwarding-stream-0_0', producerId=0, producerEpoch=3, committed=true) to node datorn:9092 (id: 0 rack: null) with correlation ID 68 TRACE 10:23:18.640[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:23:18.640[main] TestProducer - Sent record WARN 10:23:18.640[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:23:18.640[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:18.641[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:23:18.642[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 54 TRACE 10:23:18.644[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms DEBUG 10:23:18.645[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sending transactional request AddPartitionsToTxnRequestData(transactionalId='forwarding-stream-0_0', producerId=0, producerEpoch=3, topics=[AddPartitionsToTxnTopic(name='output-topic', partitions=[0])]) to node datorn:9092 (id: 0 rack: null) with correlation ID 69 WARN 10:23:18.646[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:23:18.647[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 51 being sent to partition output-topic-0 TRACE 10:23:18.647[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:18.647[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:23:18.648[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 70 TRACE 10:23:19.640[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:23:19.641[main] TestProducer - Sent record WARN 10:23:19.641[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:23:19.641[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:19.641[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:23:19.643[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 55 TRACE 10:23:19.645[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:23:19.645[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:23:19.645[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 52 being sent to partition output-topic-0 TRACE 10:23:19.645[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:19.646[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:23:19.647[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 71 TRACE 10:23:20.641[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:23:20.641[main] TestProducer - Sent record WARN 10:23:20.641[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:23:20.642[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:20.642[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:23:20.644[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 56 TRACE 10:23:20.645[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:23:20.645[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:23:20.646[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 53 being sent to partition output-topic-0 TRACE 10:23:20.646[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:20.646[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:23:20.648[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 72 TRACE 10:23:21.641[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:23:21.642[main] TestProducer - Sent record WARN 10:23:21.642[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:23:21.642[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:21.642[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:23:21.644[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 57 TRACE 10:23:21.646[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:23:21.646[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:23:21.646[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 54 being sent to partition output-topic-0 TRACE 10:23:21.647[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:21.647[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:23:21.648[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 73 TRACE 10:23:22.642[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:23:22.642[main] TestProducer - Sent record WARN 10:23:22.642[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:23:22.643[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:22.643[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:23:22.644[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 58 TRACE 10:23:22.647[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:23:22.647[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:23:22.647[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 55 being sent to partition output-topic-0 TRACE 10:23:22.647[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:22.648[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:23:22.649[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 74 TRACE 10:23:23.642[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:23:23.643[main] TestProducer - Sent record WARN 10:23:23.643[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:23:23.643[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:23.643[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:23:23.645[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 59 TRACE 10:23:23.647[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:23:23.647[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:23:23.647[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 56 being sent to partition output-topic-0 TRACE 10:23:23.647[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:23.648[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:23:23.649[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 75 TRACE 10:23:24.643[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:23:24.643[main] TestProducer - Sent record WARN 10:23:24.643[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:23:24.644[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:24.644[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:23:24.645[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 60 TRACE 10:23:24.647[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:23:24.647[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:23:24.647[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 57 being sent to partition output-topic-0 TRACE 10:23:24.647[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:24.648[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:23:24.649[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 76 TRACE 10:23:25.643[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:23:25.644[main] TestProducer - Sent record WARN 10:23:25.644[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:23:25.644[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:25.644[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:23:25.646[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 61 TRACE 10:23:25.648[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:23:25.648[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:23:25.648[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 58 being sent to partition output-topic-0 TRACE 10:23:25.649[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:25.649[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:23:25.650[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 77 TRACE 10:23:26.644[main] RecordAccumulator - [Producer clientId=producer-1] Allocating a new 16384 byte message buffer for topic input-topic partition 0 with remaining timeout 60000ms INFO 10:23:26.644[main] TestProducer - Sent record WARN 10:23:26.644[kafka-producer-network-thread | producer-1] RecordAccumulator - [Producer clientId=producer-1] Batch is sendable because full=false || expired=true || exhausted=false || closed=false || flushInProgress=false TRACE 10:23:26.645[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:26.645[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Sent produce request to 0: (type=ProduceRequest, acks=1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='' TRACE 10:23:26.646[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Received produce response from node 0 with correlation id 62 TRACE 10:23:26.648[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Allocating a new 102400 byte message buffer for topic output-topic partition 0 with remaining timeout 60000ms WARN 10:23:26.649[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Batch is sendable because full=false || expired=false || exhausted=false || closed=false || flushInProgress=true DEBUG 10:23:26.649[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Assigned producerId 0 and producerEpoch 3 to batch with base sequence 59 being sent to partition output-topic-0 TRACE 10:23:26.649[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Nodes with data ready to send: [datorn:9092 (id: 0 rack: null)] TRACE 10:23:26.649[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sent produce request to 0: (type=ProduceRequest, acks=-1, timeout=30000, partitionRecords=([PartitionProduceData(index=0, records=MemoryRecords(size=76, buffer=java.nio.HeapByteBuffer[pos=0 lim=76 cap=76]))]), transactionalId='forwarding-stream-0_0' TRACE 10:23:26.651[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Received produce response from node 0 with correlation id 78 INFO 10:23:27.644[main] KafkaProducer - [Producer clientId=producer-1] Closing the Kafka producer with timeoutMillis = 9223372036854775807 ms. DEBUG 10:23:27.645[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Beginning shutdown of Kafka producer I/O thread, sending remaining records. DEBUG 10:23:27.652[kafka-producer-network-thread | producer-1] Sender - [Producer clientId=producer-1] Shutdown of Kafka producer I/O thread has completed. INFO 10:23:27.652[main] Metrics - Metrics scheduler closed INFO 10:23:27.652[main] Metrics - Closing reporter org.apache.kafka.common.metrics.JmxReporter INFO 10:23:27.653[main] Metrics - Metrics reporters closed INFO 10:23:27.653[main] AppInfoParser - App info kafka.producer for producer-1 unregistered INFO 10:23:27.653[main] KafkaStreams - stream-client [forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c] State transition from RUNNING to PENDING_SHUTDOWN INFO 10:23:27.655[kafka-streams-close-thread] StreamThread - stream-thread [forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] Informed to shut down INFO 10:23:27.655[kafka-streams-close-thread] StreamThread - stream-thread [forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] State transition from RUNNING to PENDING_SHUTDOWN INFO 10:23:27.751[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] StreamThread - stream-thread [forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] Thread state is already PENDING_SHUTDOWN, skipping the run once call after poll request INFO 10:23:27.751[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] StreamThread - stream-thread [forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] Shutting down WARN 10:23:27.752[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] beginFlush(): 1 -> 2 java.lang.Exception: null at org.apache.kafka.clients.producer.internals.RecordAccumulator.beginFlush(RecordAccumulator.java:701) at org.apache.kafka.clients.producer.KafkaProducer.flush(KafkaProducer.java:1123) at org.apache.kafka.streams.processor.internals.StreamsProducer.flush(StreamsProducer.java:316) at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.flush(RecordCollectorImpl.java:258) at org.apache.kafka.streams.processor.internals.StreamTask.prepareCommit(StreamTask.java:391) at org.apache.kafka.streams.processor.internals.TaskManager.tryCloseCleanAllActiveTasks(TaskManager.java:848) at org.apache.kafka.streams.processor.internals.TaskManager.shutdown(TaskManager.java:791) at org.apache.kafka.streams.processor.internals.StreamThread.completeShutdown(StreamThread.java:1073) at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:561) WARN 10:23:27.752[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordAccumulator - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] awaitFlushCompletion(): 2 -> 1 java.lang.Exception: null at org.apache.kafka.clients.producer.internals.RecordAccumulator.awaitFlushCompletion(RecordAccumulator.java:720) at org.apache.kafka.clients.producer.KafkaProducer.flush(KafkaProducer.java:1126) at org.apache.kafka.streams.processor.internals.StreamsProducer.flush(StreamsProducer.java:316) at org.apache.kafka.streams.processor.internals.RecordCollectorImpl.flush(RecordCollectorImpl.java:258) at org.apache.kafka.streams.processor.internals.StreamTask.prepareCommit(StreamTask.java:391) at org.apache.kafka.streams.processor.internals.TaskManager.tryCloseCleanAllActiveTasks(TaskManager.java:848) at org.apache.kafka.streams.processor.internals.TaskManager.shutdown(TaskManager.java:791) at org.apache.kafka.streams.processor.internals.StreamThread.completeShutdown(StreamThread.java:1073) at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:561) DEBUG 10:23:27.753[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sending transactional request AddOffsetsToTxnRequestData(transactionalId='forwarding-stream-0_0', producerId=0, producerEpoch=3, groupId='forwarding-stream') to node datorn:9092 (id: 0 rack: null) with correlation ID 79 DEBUG 10:23:27.755[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sending transactional request TxnOffsetCommitRequestData(transactionalId='forwarding-stream-0_0', groupId='forwarding-stream', producerId=0, producerEpoch=3, generationId=5, memberId='forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-consumer-43fb25e4-449c-408d-bd93-6e56f83ad410', groupInstanceId=null, topics=[TxnOffsetCommitRequestTopic(name='input-topic', partitions=[TxnOffsetCommitRequestPartition(partitionIndex=0, committedOffset=214, committedLeaderEpoch=-1, committedMetadata='AQAAAXnGq560')])]) to node datorn:9092 (id: 0 rack: null) with correlation ID 80 DEBUG 10:23:27.757[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Sending transactional request EndTxnRequestData(transactionalId='forwarding-stream-0_0', producerId=0, producerEpoch=3, committed=true) to node datorn:9092 (id: 0 rack: null) with correlation ID 81 INFO 10:23:27.763[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] StreamTask - stream-thread [forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] task [0_0] Suspended running INFO 10:23:27.765[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] KafkaConsumer - [Consumer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-restore-consumer, groupId=null] Unsubscribed all topics or patterns and assigned partitions INFO 10:23:27.766[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] RecordCollectorImpl - stream-thread [forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] task [0_0] Closing record collector clean INFO 10:23:27.766[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] StreamTask - stream-thread [forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] task [0_0] Closed clean INFO 10:23:27.767[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] KafkaProducer - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Closing the Kafka producer with timeoutMillis = 9223372036854775807 ms. DEBUG 10:23:27.767[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Beginning shutdown of Kafka producer I/O thread, sending remaining records. DEBUG 10:23:27.769[kafka-producer-network-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer] Sender - [Producer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer, transactionalId=forwarding-stream-0_0] Shutdown of Kafka producer I/O thread has completed. INFO 10:23:27.769[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] Metrics - Metrics scheduler closed INFO 10:23:27.769[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] Metrics - Closing reporter org.apache.kafka.common.metrics.JmxReporter INFO 10:23:27.769[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] Metrics - Metrics reporters closed INFO 10:23:27.769[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] AppInfoParser - App info kafka.producer for forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-0_0-producer unregistered INFO 10:23:27.770[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] KafkaConsumer - [Consumer clientId=forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-restore-consumer, groupId=null] Unsubscribed all topics or patterns and assigned partitions INFO 10:23:27.771[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] Metrics - Metrics scheduler closed INFO 10:23:27.771[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] Metrics - Closing reporter org.apache.kafka.common.metrics.JmxReporter INFO 10:23:27.771[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] Metrics - Metrics reporters closed INFO 10:23:27.773[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] AppInfoParser - App info kafka.consumer for forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-consumer unregistered INFO 10:23:27.773[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] Metrics - Metrics scheduler closed INFO 10:23:27.773[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] Metrics - Closing reporter org.apache.kafka.common.metrics.JmxReporter INFO 10:23:27.773[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] Metrics - Metrics reporters closed INFO 10:23:27.774[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] AppInfoParser - App info kafka.consumer for forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1-restore-consumer unregistered INFO 10:23:27.775[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] StreamThread - stream-thread [forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] State transition from PENDING_SHUTDOWN to DEAD INFO 10:23:27.775[forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] StreamThread - stream-thread [forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-StreamThread-1] Shutdown complete INFO 10:23:27.776[kafka-admin-client-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-admin] AppInfoParser - App info kafka.admin.client for forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-admin unregistered INFO 10:23:27.779[kafka-admin-client-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-admin] Metrics - Metrics scheduler closed INFO 10:23:27.779[kafka-admin-client-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-admin] Metrics - Closing reporter org.apache.kafka.common.metrics.JmxReporter INFO 10:23:27.779[kafka-admin-client-thread | forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c-admin] Metrics - Metrics reporters closed INFO 10:23:27.779[kafka-streams-close-thread] Metrics - Metrics scheduler closed INFO 10:23:27.779[kafka-streams-close-thread] Metrics - Closing reporter org.apache.kafka.common.metrics.JmxReporter INFO 10:23:27.780[kafka-streams-close-thread] Metrics - Metrics reporters closed INFO 10:23:27.780[kafka-streams-close-thread] KafkaStreams - stream-client [forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c] State transition from PENDING_SHUTDOWN to NOT_RUNNING INFO 10:23:27.780[main] KafkaStreams - stream-client [forwarding-stream-d0aed0d0-6f35-46a5-a601-07f4d727339c] Streams client stopped completely BUILD SUCCESSFUL in 1m 2s 3 actionable tasks: 2 executed, 1 up-to-date