kirktrue commented on code in PR #14406: URL: https://github.com/apache/kafka/pull/14406#discussion_r1346606768
########## clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/ApplicationEventHandler.java: ########## @@ -0,0 +1,122 @@ +/* + * 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.kafka.clients.consumer.internals.events; + +import org.apache.kafka.clients.consumer.internals.DefaultBackgroundThread; +import org.apache.kafka.clients.consumer.internals.NetworkClientDelegate; +import org.apache.kafka.clients.consumer.internals.RequestManagers; +import org.apache.kafka.common.KafkaException; +import org.apache.kafka.common.internals.IdempotentCloser; +import org.apache.kafka.common.utils.LogContext; +import org.apache.kafka.common.utils.Time; +import org.apache.kafka.common.utils.Timer; +import org.slf4j.Logger; + +import java.io.Closeable; +import java.time.Duration; +import java.util.Objects; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.function.Supplier; + +/** + * An event handler that receives {@link ApplicationEvent application events} from the application thread which + * are then readable from the {@link ApplicationEventProcessor} in the background thread. + */ +public class ApplicationEventHandler implements Closeable { + + private final Logger log; + private final BlockingQueue<ApplicationEvent> applicationEventQueue; + private final DefaultBackgroundThread backgroundThread; + private final IdempotentCloser closer = new IdempotentCloser(); + + public ApplicationEventHandler(final Time time, + final LogContext logContext, + final BlockingQueue<ApplicationEvent> applicationEventQueue, + final Supplier<ApplicationEventProcessor> applicationEventProcessorSupplier, + final Supplier<NetworkClientDelegate> networkClientDelegateSupplier, + final Supplier<RequestManagers> requestManagersSupplier) { + this.log = logContext.logger(ApplicationEventHandler.class); + this.applicationEventQueue = applicationEventQueue; + this.backgroundThread = new DefaultBackgroundThread(time, + logContext, + applicationEventProcessorSupplier, + networkClientDelegateSupplier, + requestManagersSupplier); + this.backgroundThread.start(); + } + + /** + * Add an {@link ApplicationEvent} to the handler. + * + * @param event An {@link ApplicationEvent} created by the application thread + */ + public void add(final ApplicationEvent event) { + Objects.requireNonNull(event, "ApplicationEvent provided to add must be non-null"); + log.trace("Enqueued event: {}", event); + backgroundThread.wakeup(); + applicationEventQueue.add(event); + } + + /** + * Add a {@link CompletableApplicationEvent} to the handler. The method blocks waiting for the result, and will + * return the result value upon successful completion; otherwise throws an error. + * + * <p/> + * + * See {@link CompletableApplicationEvent#get(Timer)} and {@link Future#get(long, TimeUnit)} for more details. + * + * @param event A {@link CompletableApplicationEvent} created by the polling thread. + * @param timer Timer for which to wait for the event to complete + * @return Value that is the result of the event + * @param <T> Type of return value of the event + */ + public <T> T addAndGet(final CompletableApplicationEvent<T> event, final Timer timer) { + Objects.requireNonNull(event, "CompletableApplicationEvent provided to addAndGet must be non-null"); + Objects.requireNonNull(timer, "Timer provided to addAndGet must be non-null"); + add(event); + return event.get(timer); + } + + @Override + public void close() { + close(Duration.ofMillis(Long.MAX_VALUE)); + } + + public void close(final Duration timeout) { + Objects.requireNonNull(timeout, "Duration provided to close must be non-null"); + + closer.close( + () -> { + log.info("Closing the consumer application event handler"); + + try { + long timeoutMs = timeout.toMillis(); + + if (timeoutMs < 0) + throw new IllegalArgumentException("The timeout cannot be negative."); + + backgroundThread.close(); + log.info("The consumer application event handler was closed"); Review Comment: Changed to debug -- 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
