FrankChen021 commented on code in PR #19311: URL: https://github.com/apache/druid/pull/19311#discussion_r3259153370
########## extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/ShareGroupIndexTaskRunner.java: ########## @@ -0,0 +1,487 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.druid.indexing.kafka; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.common.annotations.VisibleForTesting; +import com.google.common.base.Supplier; +import com.google.common.collect.ImmutableMap; +import org.apache.druid.common.config.Configs; +import org.apache.druid.data.input.Committer; +import org.apache.druid.data.input.InputFormat; +import org.apache.druid.data.input.InputRow; +import org.apache.druid.data.input.InputRowSchema; +import org.apache.druid.data.input.kafka.KafkaRecordEntity; +import org.apache.druid.data.input.kafka.KafkaTopicPartition; +import org.apache.druid.indexer.TaskStatus; +import org.apache.druid.indexer.partitions.DynamicPartitionsSpec; +import org.apache.druid.indexing.appenderator.ActionBasedPublishedSegmentRetriever; +import org.apache.druid.indexing.appenderator.ActionBasedSegmentAllocator; +import org.apache.druid.indexing.common.LockGranularity; +import org.apache.druid.indexing.common.TaskLockType; +import org.apache.druid.indexing.common.TaskToolbox; +import org.apache.druid.indexing.common.actions.LockReleaseAction; +import org.apache.druid.indexing.common.actions.SegmentAllocateAction; +import org.apache.druid.indexing.common.actions.SegmentTransactionalAppendAction; +import org.apache.druid.indexing.common.actions.TaskLocks; +import org.apache.druid.indexing.common.task.InputRowFilter; +import org.apache.druid.indexing.common.task.Tasks; +import org.apache.druid.indexing.input.InputRowSchemas; +import org.apache.druid.indexing.overlord.SegmentPublishResult; +import org.apache.druid.indexing.seekablestream.SeekableStreamAppenderatorConfig; +import org.apache.druid.indexing.seekablestream.StreamChunkReader; +import org.apache.druid.indexing.seekablestream.common.AcknowledgeType; +import org.apache.druid.indexing.seekablestream.common.AcknowledgingRecordSupplier; +import org.apache.druid.indexing.seekablestream.common.OrderedPartitionableRecord; +import org.apache.druid.java.util.common.ISE; +import org.apache.druid.java.util.common.logger.Logger; +import org.apache.druid.segment.SegmentSchemaMapping; +import org.apache.druid.segment.incremental.ParseExceptionHandler; +import org.apache.druid.segment.incremental.RowIngestionMeters; +import org.apache.druid.segment.indexing.DataSchema; +import org.apache.druid.segment.realtime.SegmentGenerationMetrics; +import org.apache.druid.segment.realtime.appenderator.Appenderator; +import org.apache.druid.segment.realtime.appenderator.AppenderatorDriverAddResult; +import org.apache.druid.segment.realtime.appenderator.SegmentsAndCommitMetadata; +import org.apache.druid.segment.realtime.appenderator.StreamAppenderatorDriver; +import org.apache.druid.segment.realtime.appenderator.TransactionalSegmentPublisher; +import org.apache.druid.timeline.DataSegment; +import org.apache.druid.timeline.partition.NumberedPartialShardSpec; +import org.apache.kafka.common.errors.WakeupException; + +import javax.annotation.Nullable; +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Function; + +/** + * Single-threaded ingestion loop for {@link ShareGroupIndexTask}. Records are + * {@code ACCEPT}-acknowledged only after the containing segment is published; + * any earlier failure leaves them unacknowledged so the broker redelivers + * them once the acquisition lock expires. + */ +public class ShareGroupIndexTaskRunner +{ + private static final Logger log = new Logger(ShareGroupIndexTaskRunner.class); + private static final String SEQUENCE_NAME = "share_group_seq_0"; + + static final String METRIC_COMMIT_FAILURES = "ingest/shareGroup/commitFailures"; + + private final ShareGroupIndexTask task; + private final TaskToolbox toolbox; + private final ObjectMapper configMapper; + private final Function<ShareGroupIndexTaskIOConfig, + AcknowledgingRecordSupplier<KafkaTopicPartition, Long, KafkaRecordEntity>> supplierFactory; + + private final AtomicReference<AcknowledgingRecordSupplier<KafkaTopicPartition, Long, KafkaRecordEntity>> + activeSupplier = new AtomicReference<>(); + + ShareGroupIndexTaskRunner( + ShareGroupIndexTask task, + TaskToolbox toolbox, + ObjectMapper configMapper + ) + { + this(task, toolbox, configMapper, null); + } + + @VisibleForTesting + ShareGroupIndexTaskRunner( + ShareGroupIndexTask task, + TaskToolbox toolbox, + ObjectMapper configMapper, + @Nullable Function<ShareGroupIndexTaskIOConfig, + AcknowledgingRecordSupplier<KafkaTopicPartition, Long, KafkaRecordEntity>> supplierFactory + ) + { + this.task = task; + this.toolbox = toolbox; + this.configMapper = configMapper; + this.supplierFactory = supplierFactory != null ? supplierFactory : this::createDefaultRecordSupplier; + } + + public TaskStatus run() throws Exception + { + final DataSchema dataSchema = task.getDataSchema(); + final KafkaIndexTaskTuningConfig tuningConfig = task.getTuningConfig(); + final ShareGroupIndexTaskIOConfig ioConfig = task.getIOConfig(); + + final RowIngestionMeters rowIngestionMeters = + toolbox.getRowIngestionMetersFactory().createRowIngestionMeters(); + final ParseExceptionHandler parseExceptionHandler = new ParseExceptionHandler( + rowIngestionMeters, + tuningConfig.isLogParseExceptions(), + tuningConfig.getMaxParseExceptions(), + tuningConfig.getMaxSavedParseExceptions() + ); + final SegmentGenerationMetrics segmentGenerationMetrics = new SegmentGenerationMetrics(); + + final InputRowSchema inputRowSchema = InputRowSchemas.fromDataSchema(dataSchema); + final InputFormat inputFormat = ioConfig.getInputFormat(); + if (inputFormat == null) { + throw new ISE("inputFormat must be specified in ioConfig"); + } + + // Raw type mirrors SeekableStreamIndexTaskRunner; works around + // OrderedPartitionableRecord.getData() returning a wildcard list. + @SuppressWarnings({"rawtypes", "unchecked"}) + final StreamChunkReader chunkReader = new StreamChunkReader<KafkaRecordEntity>( + inputFormat, + inputRowSchema, + dataSchema.getTransformSpec(), + toolbox.getIndexingTmpDir(), + InputRowFilter.allowAll(), + rowIngestionMeters, + parseExceptionHandler + ); + + final LockGranularity lockGranularity = Configs.valueOrDefault( + task.getContextValue(Tasks.FORCE_TIME_CHUNK_LOCK_KEY, Tasks.DEFAULT_FORCE_TIME_CHUNK_LOCK), + Tasks.DEFAULT_FORCE_TIME_CHUNK_LOCK + ) + ? LockGranularity.TIME_CHUNK + : LockGranularity.SEGMENT; + final TaskLockType lockType = TaskLocks.determineLockTypeForAppend(task.getContext()); + + final Appenderator appenderator = toolbox.getAppenderatorsManager().createRealtimeAppenderatorForTask( + toolbox.getSegmentLoaderConfig(), + task.getId(), + dataSchema, + SeekableStreamAppenderatorConfig.fromTuningConfig( + tuningConfig.withBasePersistDirectory(toolbox.getPersistDir()), + toolbox.getProcessingConfig() + ), + toolbox.getConfig(), + segmentGenerationMetrics, + toolbox.getSegmentPusher(), + toolbox.getJsonMapper(), + toolbox.getIndexIO(), + toolbox.getIndexMerger(), + toolbox.getQueryRunnerFactoryConglomerate(), + toolbox.getSegmentAnnouncer(), + toolbox.getEmitter(), + toolbox.getQueryProcessingPool(), + toolbox.getJoinableFactory(), + toolbox.getCache(), + toolbox.getCacheConfig(), + toolbox.getCachePopulatorStats(), + toolbox.getPolicyEnforcer(), + rowIngestionMeters, + parseExceptionHandler, + toolbox.getCentralizedTableSchemaConfig(), + interval -> { + toolbox.getTaskActionClient().submit(new LockReleaseAction(interval)); + } + ); + + final StreamAppenderatorDriver driver = new StreamAppenderatorDriver( + appenderator, + new ActionBasedSegmentAllocator( + toolbox.getTaskActionClient(), + dataSchema, + (schema, row, sequenceName, previousSegmentId, skipSegmentLineageCheck) -> new SegmentAllocateAction( + schema.getDataSource(), + row.getTimestamp(), + schema.getGranularitySpec().getQueryGranularity(), + schema.getGranularitySpec().getSegmentGranularity(), + sequenceName, + previousSegmentId, + skipSegmentLineageCheck, + NumberedPartialShardSpec.instance(), + lockGranularity, + lockType + ) + ), + toolbox.getSegmentHandoffNotifierFactory(), + new ActionBasedPublishedSegmentRetriever(toolbox.getTaskActionClient()), + toolbox.getDataSegmentKiller(), + toolbox.getJsonMapper(), + segmentGenerationMetrics + ); + + final org.apache.druid.indexing.common.stats.TaskRealtimeMetricsMonitor metricsMonitor = + new org.apache.druid.indexing.common.stats.TaskRealtimeMetricsMonitor( + segmentGenerationMetrics, + rowIngestionMeters, + task.getMetricBuilder() + ); + toolbox.addMonitor(metricsMonitor); + + try (final AcknowledgingRecordSupplier<KafkaTopicPartition, Long, KafkaRecordEntity> recordSupplier = + supplierFactory.apply(ioConfig)) { + activeSupplier.set(recordSupplier); + return runLoop(driver, appenderator, recordSupplier, chunkReader, ioConfig, tuningConfig, toolbox); + } + finally { + activeSupplier.set(null); + try { + toolbox.removeMonitor(metricsMonitor); + } + catch (Exception e) { + log.warn(e, "Exception removing TaskRealtimeMetricsMonitor; continuing teardown."); + } + try { + driver.close(); + } + catch (Exception e) { + log.warn(e, "Exception closing StreamAppenderatorDriver; continuing teardown."); + } + } + } + + @VisibleForTesting + TaskStatus runLoop( + StreamAppenderatorDriver driver, + Appenderator appenderator, + AcknowledgingRecordSupplier<KafkaTopicPartition, Long, KafkaRecordEntity> recordSupplier, + StreamChunkReader chunkReader, + ShareGroupIndexTaskIOConfig ioConfig, + KafkaIndexTaskTuningConfig tuningConfig, + TaskToolbox toolbox + ) throws Exception + { + recordSupplier.subscribe(Collections.singleton(ioConfig.getTopic())); Review Comment: [P2] Close the appenderator when setup fails After this refactor, appenderator.closeNow() only runs in the try/finally that starts after recordSupplier.subscribe(...) and driver.startJob(...). If supplier creation, subscribe, or startJob throws, run() only calls driver.close(), and BaseAppenderatorDriver.close explicitly does not close the underlying Appenderator, so failed startup can leak realtime appenderator resources. Keep the appenderator close guard in run() or include setup in runLoop's try/finally. ########## embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/ShareGroupIndexTaskJsonSubmitIT.java: ########## @@ -0,0 +1,244 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.druid.testing.embedded.indexing; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.common.base.Throwables; +import org.apache.druid.data.input.impl.CsvInputFormat; +import org.apache.druid.data.input.impl.DimensionsSpec; +import org.apache.druid.data.input.impl.TimestampSpec; +import org.apache.druid.indexer.granularity.UniformGranularitySpec; +import org.apache.druid.indexing.kafka.KafkaIndexTaskModule; +import org.apache.druid.indexing.kafka.KafkaIndexTaskTuningConfig; +import org.apache.druid.indexing.kafka.ShareGroupIndexTask; +import org.apache.druid.indexing.kafka.ShareGroupIndexTaskIOConfig; +import org.apache.druid.jackson.DefaultObjectMapper; +import org.apache.druid.java.util.common.granularity.Granularities; +import org.apache.druid.query.DruidMetrics; +import org.apache.druid.segment.indexing.DataSchema; +import org.apache.druid.testing.embedded.EmbeddedBroker; +import org.apache.druid.testing.embedded.EmbeddedCoordinator; +import org.apache.druid.testing.embedded.EmbeddedDruidCluster; +import org.apache.druid.testing.embedded.EmbeddedHistorical; +import org.apache.druid.testing.embedded.EmbeddedIndexer; +import org.apache.druid.testing.embedded.EmbeddedOverlord; +import org.apache.druid.testing.embedded.junit5.EmbeddedClusterTestBase; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; + +/** Verifies that ShareGroupIndexTask can be submitted via raw JSON through the Overlord HTTP API. */ +public class ShareGroupIndexTaskJsonSubmitIT extends EmbeddedClusterTestBase +{ + private static final long SHARE_CONSUMER_READY_DELAY_MS = 3_000L; + private static final String COL_TIMESTAMP = "__time"; + private static final String COL_ITEM = "item"; + private static final String COL_VALUE = "value"; + + private final EmbeddedCoordinator coordinator = new EmbeddedCoordinator(); + private final EmbeddedOverlord overlord = new EmbeddedOverlord(); + private final EmbeddedIndexer indexer = new EmbeddedIndexer(); + private final EmbeddedHistorical historical = new EmbeddedHistorical(); + private final EmbeddedBroker broker = new EmbeddedBroker(); + + private ShareGroupKafkaResource kafkaServer; + private final ObjectMapper mapper = new DefaultObjectMapper(); + + @Override + public EmbeddedDruidCluster createCluster() + { + kafkaServer = new ShareGroupKafkaResource(); + final EmbeddedDruidCluster cluster = EmbeddedDruidCluster.withEmbeddedDerbyAndZookeeper(); + indexer.addProperty("druid.segment.handoff.pollDuration", "PT0.1s"); + cluster.addExtension(KafkaIndexTaskModule.class) + .addResource(kafkaServer) + .useLatchableEmitter() + .useDefaultTimeoutForLatchableEmitter(30) + .addCommonProperty("druid.monitoring.emissionPeriod", "PT0.1s") + .addServer(coordinator) + .addServer(overlord) + .addServer(indexer) + .addServer(historical) + .addServer(broker); + return cluster; + } + + @Test + public void test_jsonSubmit_csvFormat_ingestsData() throws Exception + { + final String topic = dataSource + "_topic"; + kafkaServer.createTopicWithPartitions(topic, 2); + kafkaServer.setShareGroupAutoOffsetReset("json-submit-group", "earliest"); + + final ShareGroupIndexTask task = buildTask(topic, "json-submit-group"); + final JsonNode taskJson = mapper.readTree(mapper.writeValueAsString(task)); + cluster.callApi().onLeaderOverlord(o -> o.runTask(task.getId(), taskJson)); + + Thread.sleep(SHARE_CONSUMER_READY_DELAY_MS); + + final int numRecords = 8; + kafkaServer.publishRecordsToTopic(topic, generateRecords(numRecords)); + + indexer.latchableEmitter().waitForEventAggregate( + event -> event.hasMetricName("ingest/events/processed") + .hasDimension(DruidMetrics.DATASOURCE, dataSource), + agg -> agg.hasSumAtLeast(numRecords) + ); + cluster.callApi().waitForAllSegmentsToBeAvailable(dataSource, coordinator, broker); + + final long deadlineMs = System.currentTimeMillis() + 30_000L; + while (System.currentTimeMillis() < deadlineMs) { + try { + cluster.callApi().verifySqlQuery("SELECT COUNT(*) FROM %s", dataSource, String.valueOf(numRecords)); + break; + } + catch (AssertionError ignored) { + Thread.sleep(500); + } + } + + cluster.callApi().onLeaderOverlord(o -> o.cancelTask(task.getId())); + cluster.callApi().waitForTaskToFinish(task.getId(), overlord.latchableEmitter()); + } + + @Test + public void test_jsonSubmit_roundTripsPayload() throws Exception + { + final String topic = dataSource + "_roundtrip_topic"; + kafkaServer.createTopicWithPartitions(topic, 1); + kafkaServer.setShareGroupAutoOffsetReset("roundtrip-group", "earliest"); + + final ShareGroupIndexTask task = buildTask(topic, "roundtrip-group"); + final JsonNode taskJson = mapper.readTree(mapper.writeValueAsString(task)); + cluster.callApi().onLeaderOverlord(o -> o.runTask(task.getId(), taskJson)); + + Thread.sleep(SHARE_CONSUMER_READY_DELAY_MS); + + final org.apache.druid.client.indexing.TaskPayloadResponse payloadResponse = + cluster.callApi().onLeaderOverlord(o -> o.taskPayload(task.getId())); + Assertions.assertEquals(task.getId(), payloadResponse.getTask()); + + final JsonNode payloadJson = mapper.valueToTree(payloadResponse.getPayload()); + Assertions.assertEquals("share_group_index", payloadJson.path("type").asText()); Review Comment: [P2] Fix JSON-submit payload expectations This new IT serializes a real ShareGroupIndexTask and reads the Overlord payload, but the task is registered as index_kafka_share_group and serializes ioConfig at the top level, not under spec. The round-trip test therefore fails on the type assertion, and still fails on the topic/group assertions after that is fixed; the malformed-json test below also uses the wrong type/shape, so it does not exercise the intended missing-field path. ########## embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/ShareGroupIndexTaskJsonSubmitIT.java: ########## @@ -0,0 +1,244 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.druid.testing.embedded.indexing; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.common.base.Throwables; +import org.apache.druid.data.input.impl.CsvInputFormat; +import org.apache.druid.data.input.impl.DimensionsSpec; +import org.apache.druid.data.input.impl.TimestampSpec; +import org.apache.druid.indexer.granularity.UniformGranularitySpec; +import org.apache.druid.indexing.kafka.KafkaIndexTaskModule; +import org.apache.druid.indexing.kafka.KafkaIndexTaskTuningConfig; +import org.apache.druid.indexing.kafka.ShareGroupIndexTask; +import org.apache.druid.indexing.kafka.ShareGroupIndexTaskIOConfig; +import org.apache.druid.jackson.DefaultObjectMapper; +import org.apache.druid.java.util.common.granularity.Granularities; +import org.apache.druid.query.DruidMetrics; +import org.apache.druid.segment.indexing.DataSchema; +import org.apache.druid.testing.embedded.EmbeddedBroker; +import org.apache.druid.testing.embedded.EmbeddedCoordinator; +import org.apache.druid.testing.embedded.EmbeddedDruidCluster; +import org.apache.druid.testing.embedded.EmbeddedHistorical; +import org.apache.druid.testing.embedded.EmbeddedIndexer; +import org.apache.druid.testing.embedded.EmbeddedOverlord; +import org.apache.druid.testing.embedded.junit5.EmbeddedClusterTestBase; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; + +/** Verifies that ShareGroupIndexTask can be submitted via raw JSON through the Overlord HTTP API. */ +public class ShareGroupIndexTaskJsonSubmitIT extends EmbeddedClusterTestBase +{ + private static final long SHARE_CONSUMER_READY_DELAY_MS = 3_000L; + private static final String COL_TIMESTAMP = "__time"; + private static final String COL_ITEM = "item"; + private static final String COL_VALUE = "value"; + + private final EmbeddedCoordinator coordinator = new EmbeddedCoordinator(); + private final EmbeddedOverlord overlord = new EmbeddedOverlord(); + private final EmbeddedIndexer indexer = new EmbeddedIndexer(); + private final EmbeddedHistorical historical = new EmbeddedHistorical(); + private final EmbeddedBroker broker = new EmbeddedBroker(); + + private ShareGroupKafkaResource kafkaServer; + private final ObjectMapper mapper = new DefaultObjectMapper(); + + @Override + public EmbeddedDruidCluster createCluster() + { + kafkaServer = new ShareGroupKafkaResource(); + final EmbeddedDruidCluster cluster = EmbeddedDruidCluster.withEmbeddedDerbyAndZookeeper(); + indexer.addProperty("druid.segment.handoff.pollDuration", "PT0.1s"); + cluster.addExtension(KafkaIndexTaskModule.class) + .addResource(kafkaServer) + .useLatchableEmitter() + .useDefaultTimeoutForLatchableEmitter(30) + .addCommonProperty("druid.monitoring.emissionPeriod", "PT0.1s") + .addServer(coordinator) + .addServer(overlord) + .addServer(indexer) + .addServer(historical) + .addServer(broker); + return cluster; + } + + @Test + public void test_jsonSubmit_csvFormat_ingestsData() throws Exception + { + final String topic = dataSource + "_topic"; + kafkaServer.createTopicWithPartitions(topic, 2); + kafkaServer.setShareGroupAutoOffsetReset("json-submit-group", "earliest"); + + final ShareGroupIndexTask task = buildTask(topic, "json-submit-group"); + final JsonNode taskJson = mapper.readTree(mapper.writeValueAsString(task)); + cluster.callApi().onLeaderOverlord(o -> o.runTask(task.getId(), taskJson)); + + Thread.sleep(SHARE_CONSUMER_READY_DELAY_MS); + + final int numRecords = 8; + kafkaServer.publishRecordsToTopic(topic, generateRecords(numRecords)); + + indexer.latchableEmitter().waitForEventAggregate( + event -> event.hasMetricName("ingest/events/processed") + .hasDimension(DruidMetrics.DATASOURCE, dataSource), + agg -> agg.hasSumAtLeast(numRecords) + ); + cluster.callApi().waitForAllSegmentsToBeAvailable(dataSource, coordinator, broker); + + final long deadlineMs = System.currentTimeMillis() + 30_000L; Review Comment: [P3] Fail when the row-count wait expires If verifySqlQuery never observes the expected count, this loop exits when the deadline expires and the test continues to cancel the task without failing. A broken JSON-submit ingestion/query path can pass after 30 seconds. Track the last assertion and fail after the deadline, or reuse the existing assertRowCountEventually pattern. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
