wcarlson5 commented on a change in pull request #11479: URL: https://github.com/apache/kafka/pull/11479#discussion_r753367564
########## File path: streams/src/main/java/org/apache/kafka/streams/processor/internals/namedtopology/KafkaStreamsNamedTopologyWrapper.java ########## @@ -118,40 +133,90 @@ public NamedTopologyBuilder newNamedTopologyBuilder(final String topologyName) { /** * Add a new NamedTopology to a running Kafka Streams app. If multiple instances of the application are running, - * you should inform all of them by calling {@link #addNamedTopology(NamedTopology)} on each client in order for + * you should inform all of them by calling {@code #addNamedTopology(NamedTopology)} on each client in order for * it to begin processing the new topology. * * @throws IllegalArgumentException if this topology name is already in use * @throws IllegalStateException if streams has not been started or has already shut down * @throws TopologyException if this topology subscribes to any input topics or pattern already in use */ - public void addNamedTopology(final NamedTopology newTopology) { + public AddNamedTopologyResult addNamedTopology(final NamedTopology newTopology) { + log.debug("Adding topology: {}", newTopology.name()); if (hasStartedOrFinishedShuttingDown()) { throw new IllegalStateException("Cannot add a NamedTopology while the state is " + super.state); } else if (getTopologyByName(newTopology.name()).isPresent()) { throw new IllegalArgumentException("Unable to add the new NamedTopology " + newTopology.name() + " as another of the same name already exists"); } - topologyMetadata.registerAndBuildNewTopology(newTopology.internalTopologyBuilder()); + return new AddNamedTopologyResult( + topologyMetadata.registerAndBuildNewTopology(newTopology.internalTopologyBuilder()) + ); } /** * Remove an existing NamedTopology from a running Kafka Streams app. If multiple instances of the application are - * running, you should inform all of them by calling {@link #removeNamedTopology(String)} on each client to ensure + * running, you should inform all of them by calling {@code #removeNamedTopology(String)} on each client to ensure * it stops processing the old topology. * + * @param topologyToRemove name of the topology to be removed + * @param resetOffsets whether to reset the committed offsets for any source topics + * * @throws IllegalArgumentException if this topology name cannot be found * @throws IllegalStateException if streams has not been started or has already shut down * @throws TopologyException if this topology subscribes to any input topics or pattern already in use */ - public void removeNamedTopology(final String topologyToRemove) { + public RemoveNamedTopologyResult removeNamedTopology(final String topologyToRemove, final boolean resetOffsets) { + log.debug("Removing topology: {}", topologyToRemove); if (!isRunningOrRebalancing()) { throw new IllegalStateException("Cannot remove a NamedTopology while the state is " + super.state); } else if (!getTopologyByName(topologyToRemove).isPresent()) { throw new IllegalArgumentException("Unable to locate for removal a NamedTopology called " + topologyToRemove); } + final Set<TopicPartition> partitionsToReset = metadataForLocalThreads() + .stream() + .flatMap(t -> { + final HashSet<TaskMetadata> tasks = new HashSet<>(); + tasks.addAll(t.activeTasks()); + tasks.addAll(t.standbyTasks()); + return tasks.stream(); + }) + .flatMap(t -> t.topicPartitions().stream()) + .filter(t -> topologyMetadata.sourceTopologies(topologyToRemove).contains(t.topic())) + .collect(Collectors.toSet()); + + + + final KafkaFuture<Void> removeTopologyFuture = topologyMetadata.unregisterTopology(topologyToRemove); + + if (resetOffsets) { + log.info("partitions to reset: {}", partitionsToReset); + if (!partitionsToReset.isEmpty()) { + try { + removeTopologyFuture.get(); Review comment: How about something like this? ``` if (resetOffsets) { KafkaFutureImpl<Void> future = new KafkaFutureImpl<>(); log.info("partitions to reset: {}", partitionsToReset); if (!partitionsToReset.isEmpty()) { removeTopologyFuture.whenComplete( (v, e) -> { DeleteConsumerGroupOffsetsResult deleteOffsetsResult = null; while (deleteOffsetsResult == null) { try { deleteOffsetsResult = adminClient.deleteConsumerGroupOffsets( applicationConfigs.getString(StreamsConfig.APPLICATION_ID_CONFIG), partitionsToReset); deleteOffsetsResult.all().get(); } catch (InterruptedException ex) { ex.printStackTrace(); break; } catch (ExecutionException ex) { ex.printStackTrace(); } try { Thread.sleep(100); } catch (InterruptedException ex) { ex.printStackTrace(); } } future.complete(null); }); return new RemoveNamedTopologyResult(removeTopologyFuture, future); } } ``` -- 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: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org