diff --git a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/compact/KafkaIngestionWithCompactionTest.java b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/compact/KafkaIngestionWithCompactionTest.java new file mode 100644 index 000000000000..1b11ffd9f84b --- /dev/null +++ b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/compact/KafkaIngestionWithCompactionTest.java @@ -0,0 +1,473 @@ +/* + * 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.compact; + +import org.apache.druid.catalog.guice.CatalogClientModule; +import org.apache.druid.catalog.guice.CatalogCoordinatorModule; +import org.apache.druid.common.utils.IdUtils; +import org.apache.druid.data.input.impl.DimensionsSpec; +import org.apache.druid.data.input.impl.StringDimensionSchema; +import org.apache.druid.data.input.impl.TimestampSpec; +import org.apache.druid.indexer.CompactionEngine; +import org.apache.druid.indexer.granularity.UniformGranularitySpec; +import org.apache.druid.indexer.partitions.DimensionRangePartitionsSpec; +import org.apache.druid.indexer.partitions.PartitionsSpec; +import org.apache.druid.indexing.compact.CompactionSupervisorSpec; +import org.apache.druid.indexing.kafka.simulate.KafkaResource; +import org.apache.druid.indexing.kafka.supervisor.KafkaSupervisorSpec; +import org.apache.druid.indexing.kafka.supervisor.KafkaSupervisorSpecBuilder; +import org.apache.druid.indexing.overlord.Segments; +import org.apache.druid.jackson.DefaultObjectMapper; +import org.apache.druid.java.util.common.HumanReadableBytes; +import org.apache.druid.java.util.common.Numbers; +import org.apache.druid.java.util.common.granularity.Granularities; +import org.apache.druid.java.util.common.granularity.Granularity; +import org.apache.druid.query.DruidMetrics; +import org.apache.druid.rpc.UpdateResponse; +import org.apache.druid.segment.metadata.DefaultIndexingStateFingerprintMapper; +import org.apache.druid.segment.metadata.IndexingStateCache; +import org.apache.druid.segment.metadata.IndexingStateFingerprintMapper; +import org.apache.druid.server.compaction.CompactionCandidateSearchPolicy; +import org.apache.druid.server.compaction.MostFragmentedIntervalFirstPolicy; +import org.apache.druid.server.coordinator.ClusterCompactionConfig; +import org.apache.druid.server.coordinator.DataSourceCompactionConfig; +import org.apache.druid.server.coordinator.InlineSchemaDataSourceCompactionConfig; +import org.apache.druid.server.coordinator.UserCompactionTaskDimensionsConfig; +import org.apache.druid.server.coordinator.UserCompactionTaskGranularityConfig; +import org.apache.druid.server.coordinator.UserCompactionTaskIOConfig; +import org.apache.druid.server.coordinator.UserCompactionTaskQueryTuningConfig; +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.EmbeddedRouter; +import org.apache.druid.testing.embedded.indexing.MoreResources; +import org.apache.druid.testing.embedded.junit5.EmbeddedClusterTestBase; +import org.apache.druid.testing.embedded.tools.EventSerializer; +import org.apache.druid.testing.embedded.tools.JsonEventSerializer; +import org.apache.druid.testing.embedded.tools.StreamGenerator; +import org.apache.druid.testing.embedded.tools.WikipediaStreamEventStreamGenerator; +import org.apache.druid.client.BrokerServerView; +import org.apache.druid.client.selector.ServerSelector; +import org.apache.druid.query.TableDataSource; +import org.apache.druid.timeline.DataSegment; +import org.apache.druid.timeline.SegmentId; +import org.apache.druid.timeline.VersionedIntervalTimeline; +import org.apache.druid.timeline.partition.DimensionRangeShardSpec; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.hamcrest.Matchers; +import org.joda.time.Period; +import org.junit.Assume; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.MethodSource; + +import javax.annotation.Nullable; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.stream.Collectors; + +/** + * Tests the edge case of Kafka ingestion and compaction (including minor compaction) + * running concurrently on the same datasource and interval. + */ +public class KafkaIngestionWithCompactionTest extends EmbeddedClusterTestBase +{ + private final KafkaResource kafkaServer = new KafkaResource(); + private final EmbeddedBroker broker = new EmbeddedBroker(); + private final EmbeddedIndexer indexer = new EmbeddedIndexer() + .setServerMemory(2_000_000_000L) + .addProperty("druid.worker.capacity", "20"); + private final EmbeddedOverlord overlord = new EmbeddedOverlord() + .addProperty("druid.manager.segments.pollDuration", "PT1s") + .addProperty("druid.manager.segments.useIncrementalCache", "always"); + private final EmbeddedHistorical historical = new EmbeddedHistorical(); + private final EmbeddedCoordinator coordinator = new EmbeddedCoordinator() + .addProperty("druid.manager.segments.useIncrementalCache", "always"); + + @Override + public EmbeddedDruidCluster createCluster() + { + return EmbeddedDruidCluster.withEmbeddedDerbyAndZookeeper() + .useLatchableEmitter() + .useDefaultTimeoutForLatchableEmitter(600) + .addCommonProperty("druid.auth.authorizers", "[\"allowAll\"]") + .addCommonProperty("druid.auth.authorizer.allowAll.type", "allowAll") + .addCommonProperty("druid.auth.authorizer.allowAll.policy.type", "noRestriction") + .addCommonProperty( + "druid.policy.enforcer.allowedPolicies", + "[\"org.apache.druid.query.policy.NoRestrictionPolicy\"]" + ) + .addCommonProperty("druid.policy.enforcer.type", "restrictAllTables") + .addExtensions(CatalogClientModule.class, CatalogCoordinatorModule.class) + .addResource(kafkaServer) + .addServer(coordinator) + .addServer(overlord) + .addServer(indexer) + .addServer(historical) + .addServer(broker) + .addServer(new EmbeddedRouter()); + } + + /** + * Tests minor compaction while a slow Kafka supervisor holds an append lock on the same interval. + *

+ * Setup: + * - fastSupervisor: short task duration (3s), commits segments quickly + * - slowSupervisor: long task duration (100s), holds an append lock throughout the test + * and never commits its segment during the test window + *

+ * Expected behaviour: + * Step 1: Ingest 1000 rows via fastSupervisor → 1 DAY segment at V0, row count = 1000 + * Step 2: Full compaction → 1 compacted DAY segment at V1, row count = 1000 + * Step 3: Ingest 1000 more rows via fastSupervisor + 20 rows via slowSupervisor (stays realtime) + * → new published segment appended at V1, realtime segment visible but not committed + * Step 4: Minor compaction triggers (new uncompacted segment pushes threshold) → 2 compacted DAY + * segments at V2, row count = 2000 (slowSupervisor's 20 rows are pending upgrade). + * After slow supervisor publishes 30 more rows, total reaches 2050. + */ + @MethodSource("getPolicyAndPartition") + @ParameterizedTest(name = "policy={0}, partitionsSpec={1}") + public void test_minorCompaction_withConcurrentKafkaIngestion( + MostFragmentedIntervalFirstPolicy policy, + PartitionsSpec partitionsSpec + ) + { + final String topic = IdUtils.getRandomId(); + final String slowTopic = topic + "_slow"; + configureCompaction(CompactionEngine.MSQ, policy); + + final KafkaSupervisorSpecBuilder baseSupervisorSpecBuilder = MoreResources.Supervisor.KAFKA_JSON + .get() + .withDataSchema(schema -> schema.withTimestamp(new TimestampSpec("timestamp", "iso", null)) + .withDimensions(DimensionsSpec.builder().useSchemaDiscovery(true).build()) + .withGranularity(new UniformGranularitySpec(Granularities.DAY, null, null))) + .withTuningConfig(tuningConfig -> tuningConfig.withMaxRowsPerSegment(10_000)) + .withIoConfig(ioConfig -> ioConfig.withConsumerProperties(kafkaServer.consumerProperties()).withTaskCount(1)) + .withContext(Map.of("useConcurrentLocks", true)); + + final KafkaSupervisorSpec fastSupervisor = baseSupervisorSpecBuilder + .withIoConfig(ioConfig -> ioConfig.withTaskDuration(Period.millis(3_000))) + .withId(topic) + .build(dataSource, topic); + cluster.callApi().postSupervisor(fastSupervisor); + + final KafkaSupervisorSpec slowSupervisor = baseSupervisorSpecBuilder + .withIoConfig(ioConfig -> ioConfig.withTaskDuration(Period.millis(100_000))) + .withId(slowTopic) + .build(dataSource, slowTopic); + cluster.callApi().postSupervisor(slowSupervisor); + + // Step 1: ingest 1000 rows and verify 1000 rows, first set of segments published as V0 (epoch time) + System.out.println("===== Step 1: start"); + long ingestionPersistedCount = getTotalPersistCount(); + ingestRecords(topic, 1000); + waitUntilPublishedRecordsAreIngested(ingestionPersistedCount); + ingestionPersistedCount = getTotalPersistCount(); + waitForUsedSegmentsOnBroker(); + + Assertions.assertEquals(1000, getTotalRowCount()); + + // Step 2: full compaction → 1 compacted segment, second set of segments published as V1 + System.out.println("===== Step 2: start"); + InlineSchemaDataSourceCompactionConfig dayGranularityConfig = + InlineSchemaDataSourceCompactionConfig + .builder() + .forDataSource(dataSource) + .withSkipOffsetFromLatest(Period.seconds(0)) + .withGranularitySpec(new UserCompactionTaskGranularityConfig(Granularities.DAY, null, false)) + .withDimensionsSpec(new UserCompactionTaskDimensionsConfig( + WikipediaStreamEventStreamGenerator.dimensions() + .stream() + .map(StringDimensionSchema::new) + .collect(Collectors.toUnmodifiableList()))) + .withTaskContext(Map.of("useConcurrentLocks", true)) + .withIoConfig(new UserCompactionTaskIOConfig(true)) + .withTuningConfig(UserCompactionTaskQueryTuningConfig.builder().partitionsSpec(partitionsSpec).build()) + .build(); + + int finishedCompactionTaskCount = getFinishedCompactionTaskCount(); + runCompactionWithSpec(dayGranularityConfig); + waitForNextCompactionTaskToFinish(finishedCompactionTaskCount); + finishedCompactionTaskCount = getFinishedCompactionTaskCount(); + waitForUsedSegmentsOnBroker(); + + // verify only 1 compacted segment with 1000 rows, all historical segments at a single version + Set compactedSegmentsStep2 = overlord.bindings() + .segmentsMetadataStorage() + .retrieveAllUsedSegments(dataSource, Segments.ONLY_VISIBLE) + .stream() + .filter(s -> !s.isTombstone()) + .filter(s -> s.getLastCompactionState() != null) + .collect(Collectors.toSet()); + Assertions.assertEquals(1, compactedSegmentsStep2.size()); + Assertions.assertEquals(1000, getTotalRowCount()); + final String V1 = logActiveSegmentsAndReturnVersion(); + + // verify compaction queue is empty + overlord.latchableEmitter().waitForNextEvent(event -> event.hasMetricName("segment/metadataCache/sync/time")); + overlord.latchableEmitter().waitForNextEvent(event -> event.hasMetricName("compact/runScheduler/time")); + Assertions.assertEquals(0L, overlord.latchableEmitter().getMetricValues("interval/waitCompact/count", Map.of(DruidMetrics.DATASOURCE, dataSource)).getLast()); + + // Step 3: ingest 1000 more rows via supervisor1 + 20 rows via slowSupervisor (holds append lock, never commits), append segment to V1 + System.out.println("===== Step 3: start"); + ingestRecords(slowTopic, 20); // the slowTopic is consumed by slowSupervisor, which stays as realtime segment for our testing purpose + ingestRecords(topic, 1000); + waitUntilPublishedRecordsAreIngested(ingestionPersistedCount); + // sometimes this fails when compaction task finishes too fast and we're already on V2..... + Assertions.assertEquals(V1, logActiveSegmentsAndReturnVersion()); + + // Step 4: wait for minor compaction to trigger and finish, third set of segments published as V2 + System.out.println("===== Step 4: start"); + waitForNextCompactionTaskToFinish(finishedCompactionTaskCount); // compaction task triggered by the new segment published in step 3 + + // verify compaction queue is empty + overlord.latchableEmitter().waitForNextEvent(event -> event.hasMetricName("segment/metadataCache/sync/time")); + overlord.latchableEmitter().waitForNextEvent(event -> event.hasMetricName("compact/runScheduler/time")); + Assertions.assertEquals(0L, overlord.latchableEmitter().getMetricValues("interval/waitCompact/count", Map.of(DruidMetrics.DATASOURCE, dataSource)).getLast()); + + // Get the compacted segments from overlord and verify their shardspecs + Set compactedSegments = overlord.bindings() + .segmentsMetadataStorage() + .retrieveAllUsedSegments(dataSource, Segments.ONLY_VISIBLE) + .stream() + .filter(s -> !s.isTombstone()) + .filter(s -> s.getLastCompactionState() != null) + .collect(Collectors.toSet()); + + // 2 segments: 1 originally compacted (2000 rows) + 1 minor-compacted (1000 from fastSupervisor + 20 from slowSupervisor) + Assertions.assertEquals(2, compactedSegments.size()); + Assertions.assertEquals(2000, compactedSegments.stream().mapToLong(DataSegment::getTotalRows).sum()); + + // Wait until the broker's timeline contains the most recent version + Set expectedIds = compactedSegments.stream().map(DataSegment::getId).collect(Collectors.toSet()); + waitForSegmentIdsOnBroker(expectedIds); + final String V2 = logActiveSegmentsAndReturnVersion(); + + if (getTotalRowCount() == 2020) { + // not sure why it hits this path.... but when it does, we can get to the expected row count + logActiveSegmentsAndReturnVersion(); + } + // we hit this path if pending segments are not upgraded successfully, and the totalRowCount would be 2000 until the slowSupervisor publishes its segment + Assume.assumeTrue(getTotalRowCount() == 2000); + + ingestRecords(slowTopic, 30); + waitForRowCountAndVersion(2050, V2); + Assertions.assertEquals(2000, getTotalRowCount()); // we never saw the whole 2050 records + } + + private void configureCompaction( + CompactionEngine compactionEngine, + @Nullable CompactionCandidateSearchPolicy policy + ) + { + final UpdateResponse updateResponse = cluster.callApi().onLeaderOverlord( + o -> o.updateClusterCompactionConfig(new ClusterCompactionConfig( + 1.0, + 100, + policy, + true, + compactionEngine, + true + )) + ); + Assertions.assertTrue(updateResponse.isSuccess()); + } + + private void ingestRecords(String targetTopic, int totalRows) + { + final EventSerializer serializer = new JsonEventSerializer(overlord.bindings().jsonMapper()); + final StreamGenerator streamGenerator = new WikipediaStreamEventStreamGenerator( + serializer, totalRows, 1_000 + ); + List records = streamGenerator.generateEvents(1); + + ArrayList> producerRecords = new ArrayList<>(); + for (byte[] record : records) { + producerRecords.add(new ProducerRecord<>(targetTopic, record)); + } + CompletableFuture.runAsync(() -> kafkaServer.produceRecordsToTopic(producerRecords)); + } + + private long getTotalPersistCount() + { + return indexer.latchableEmitter() + .getMetricValues("ingest/persists/count", Map.of(DruidMetrics.DATASOURCE, dataSource)) + .stream() + .mapToLong(Number::longValue) + .sum(); + } + + private void waitUntilPublishedRecordsAreIngested(long previousPersistTotal) + { + indexer.latchableEmitter().waitForEventAggregate( + event -> event.hasMetricName("ingest/persists/count") + .hasDimension(DruidMetrics.DATASOURCE, dataSource) + .hasValueMatching(Matchers.greaterThan(0L)), + agg -> agg.hasSumAtLeast(previousPersistTotal + 1) + ); + System.out.println("===== persist detected for datasource " + dataSource); + } + + /** + * Logs active segments and asserts all rows share a single version. + * Each CSV line has the form: version,type,segment_id,shard_spec,num_rows + */ + private String logActiveSegmentsAndReturnVersion() + { + final String result = cluster.runSql( + "SELECT version, CASE WHEN is_realtime=1 THEN 'realtime' ELSE 'historical' END AS type," + + " segment_id, shard_spec, num_rows" + + " FROM sys.segments WHERE is_active=1 AND datasource='%s'" + + " ORDER BY version desc, segment_id", + dataSource + ); + System.out.println("===== active segments: \n" + result + "\n" + "===== end active segments"); + Set versions = Arrays.stream(result.split("\n")) + .map(String::trim) + .filter(s -> !s.isEmpty()) + .map(line -> line.split(",", 2)[0]) + .collect(Collectors.toSet()); + return versions.iterator().next(); + } + + private int getTotalRowCount() + { + return Numbers.parseInt(cluster.runSql("SELECT COUNT(*) as cnt FROM \"%s\"", dataSource)); + } + + private void waitForRowCountAndVersion(int expectedRowCount, String expectedVersion) + { + long startMs = System.currentTimeMillis(); + long deadlineMs = startMs + 60_000; + while (System.currentTimeMillis() < deadlineMs) { + int count = getTotalRowCount(); + long elapsedMs = System.currentTimeMillis() - startMs; + System.out.println("===== waiting for row count " + expectedRowCount + ", current=" + count + " elapsed=" + elapsedMs + "ms"); + if (count == expectedRowCount) { + Assertions.assertEquals(expectedVersion, logActiveSegmentsAndReturnVersion()); + System.out.println("===== row count reached " + expectedRowCount + " after " + elapsedMs + "ms"); + return; + } + try { + Thread.sleep(1_000); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException(e); + } + } + long elapsedMs = System.currentTimeMillis() - startMs; + Assertions.assertEquals(expectedVersion, logActiveSegmentsAndReturnVersion()); + System.out.println("===== timed out waiting for row count " + expectedRowCount + " after " + elapsedMs + "ms"); + } + + private void waitForUsedSegmentsOnBroker() + { + Set usedSegmentIds = coordinator.bindings() + .segmentsMetadataStorage() + .retrieveAllUsedSegments(dataSource, Segments.INCLUDING_OVERSHADOWED) + .stream() + .map(DataSegment::getId) + .collect(Collectors.toSet()); + waitForSegmentIdsOnBroker(usedSegmentIds); + } + + private void waitForSegmentIdsOnBroker(Set expectedIds) + { + final BrokerServerView brokerView = broker.bindings().getInstance(BrokerServerView.class); + final TableDataSource tableDataSource = new TableDataSource(dataSource); + + long startMs = System.currentTimeMillis(); + long deadlineMs = startMs + 60_000; + while (System.currentTimeMillis() < deadlineMs) { + Optional> timeline = brokerView.getTimeline(tableDataSource); + Set brokerIds = timeline + .map(t -> t.iterateAllObjects().stream() + .map(selector -> selector.getSegment().getId()) + .collect(Collectors.toSet())) + .orElse(Set.of()); + + long elapsedMs = System.currentTimeMillis() - startMs; + System.out.println("===== waiting for segment IDs on broker elapsed=" + elapsedMs + "ms" + + " have=" + brokerIds.size() + " of " + expectedIds.size() + + " missing=" + expectedIds.stream() + .filter(id -> !brokerIds.contains(id)) + .collect(Collectors.toSet())); + if (brokerIds.containsAll(expectedIds)) { + System.out.println("===== all " + expectedIds.size() + " segment IDs present on broker after " + elapsedMs + "ms"); + return; + } + try { + Thread.sleep(1_000); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException(e); + } + } + Assertions.fail("Timed out waiting for segment IDs " + expectedIds + " to appear on broker"); + } + + private void runCompactionWithSpec(DataSourceCompactionConfig config) + { + cluster.callApi().postSupervisor(new CompactionSupervisorSpec(config, false, null)); + } + + private int getFinishedCompactionTaskCount() + { + return (int) overlord.latchableEmitter() + .getMetricEvents("task/run/time") + .stream() + .filter(e -> { + Object taskType = e.getUserDims().get(DruidMetrics.TASK_TYPE); + Object ds = e.getUserDims().get(DruidMetrics.DATASOURCE); + return dataSource.equals(ds) && "compact".equals(taskType); + }) + .count(); + } + + private void waitForNextCompactionTaskToFinish(int previousCount) + { + overlord.latchableEmitter().waitForEventAggregate( + event -> event.hasMetricName("task/run/time") + .hasDimension(DruidMetrics.TASK_TYPE, "compact") + .hasDimension(DruidMetrics.DATASOURCE, dataSource), + agg -> agg.hasCountAtLeast(previousCount + 1) + ); + } + + public static List getPolicyAndPartition() + { + return List.of( + new Object[]{ + new MostFragmentedIntervalFirstPolicy(1, new HumanReadableBytes("100KiB"), null, 80, null, null), + new DimensionRangePartitionsSpec(null, 10_000, List.of("page"), false) + } + ); + } +} \ No newline at end of file diff --git a/server/src/main/java/org/apache/druid/client/BrokerServerView.java b/server/src/main/java/org/apache/druid/client/BrokerServerView.java index 61a85cd4c2ce..111ae7ae07e2 100644 --- a/server/src/main/java/org/apache/druid/client/BrokerServerView.java +++ b/server/src/main/java/org/apache/druid/client/BrokerServerView.java @@ -276,6 +276,12 @@ private void serverAddedSegment(final DruidServerMetadata server, final DataSegm // loop... if (!server.getType().equals(ServerType.BROKER)) { log.debug("Adding segment[%s] for server[%s]", segment, server); + if (server.getType().equals(ServerType.INDEXER_EXECUTOR)) { + System.out.println("==== BrokerServerView: realtime segment added" + + " segmentId=" + segmentId + + " version=" + segment.getVersion() + + " server=" + server.getName()); + } ServerSelector selector = selectors.get(segmentId); if (selector == null) { selector = new ServerSelector(segment, historicalTierSelectorStrategy, realtimeTierSelectorStrategy, brokerViewOfCoordinatorConfig); diff --git a/server/src/main/java/org/apache/druid/client/CachingClusteredClient.java b/server/src/main/java/org/apache/druid/client/CachingClusteredClient.java index 9305b1b88e9f..faba94e37695 100644 --- a/server/src/main/java/org/apache/druid/client/CachingClusteredClient.java +++ b/server/src/main/java/org/apache/druid/client/CachingClusteredClient.java @@ -345,6 +345,19 @@ ClusterQueryResult run( } final Set segmentServers = computeSegmentsToQuery(timeline, specificSegments); + System.out.println("==== CachingClusteredClient: query[" + query.getId() + "] segments=" + segmentServers.size() + + " versions=" + segmentServers.stream() + .filter(s -> s.getServer() != null) + .map(s -> s.getServer().getSegment().getVersion()) + .distinct() + .sorted() + .collect(Collectors.joining(","))); + log.infoSegmentIds( + segmentServers.stream() + .filter(s -> s.getServer() != null) + .map(s -> s.getServer().getSegment().getId()), + StringUtils.format("Query[%s] found segments from timeline lookup", query.getId()) + ); final CloneQueryMode cloneQueryMode = query.context().getCloneQueryMode(); @Nullable final byte[] queryCacheKey = cacheKeyManager.computeSegmentLevelQueryCacheKey(); diff --git a/server/src/main/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderator.java b/server/src/main/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderator.java index 94e7a56679b1..af6707fb2b1d 100644 --- a/server/src/main/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderator.java +++ b/server/src/main/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderator.java @@ -527,7 +527,11 @@ private Sink getOrCreateSink(final SegmentIdWithShardSpec identifier) addSink(identifier, retVal); try { - segmentAnnouncer.announceSegment(retVal.getSegment()); + final DataSegment announcedSegment = retVal.getSegment(); + System.out.println("==== StreamAppenderator: announcing new realtime segment" + + " segmentId=" + announcedSegment.getId() + + " version=" + announcedSegment.getVersion()); + segmentAnnouncer.announceSegment(announcedSegment); } catch (IOException e) { log.makeAlert(e, "Failed to announce new segment[%s]", schema.getDataSource()) @@ -1198,6 +1202,10 @@ public void registerUpgradedPendingSegment(PendingSegmentRecord pendingSegmentRe final DataSegment baseSegment = sink.getSegment(); final DataSegment newSegment = getUpgradedSegment(baseSegment, upgradedPendingSegment); + System.out.println("==== StreamAppenderator: announcing upgraded realtime segment" + + " segmentId=" + newSegment.getId() + + " version=" + newSegment.getVersion() + + " upgradedFrom=" + baseSegment.getVersion()); segmentAnnouncer.announceSegment(newSegment); baseSegmentToUpgradedSegments.get(basePendingSegment).add(upgradedPendingSegment); upgradedSegmentToBaseSegment.put(upgradedPendingSegment, basePendingSegment);