package org.apache.kafka.server.auditor;

import org.apache.kafka.common.Configurable;
import org.apache.kafka.common.acl.AclOperation;
import org.apache.kafka.common.annotation.InterfaceStability;
import org.apache.kafka.common.resource.ResourcePattern;
import org.apache.kafka.server.authorizer.AuthorizableRequestContext;
import org.apache.kafka.server.authorizer.AuthorizationResult;

import java.util.List;

/**
 * An auditor class that can be used to hook into the request after its completion and do auditing tasks, such as
 * logging to a special file or sending information about the request to external systems.
 * Threading model:
 * <ul>
 *     <li>The auditor implementation must be thread-safe.</li>
 *     <li>The auditor implementation is expected to be asynchronous with low latency as it is used in performance
 *     sensitive areas, such as in handling produce requests.</li>
 *     <li>Any threads or thread pools used for processing remote operations asynchronously can be started during
 *     start(). These threads must be shutdown during close().</li>
 * </ul>
 */
@InterfaceStability.Evolving
public interface Auditor extends Configurable, AutoCloseable {

    class AuditInfoAuditAuthInfo {

        private final ResourcePattern resourcePattern;
        private final AclOperation operation;
        private final AuthorizationResult allowed;
        private final int error;

        public AuditInfoAuditAuthInfo(AclOperation operation, ResourcePattern resourcePattern) {
            this.operation = operation;
            this.resourcePattern = resourcePattern;
            this.allowed = AuthorizationResult.ALLOWED;
            this.error = 0;
        }

        public AuditInfoAuditAuthInfo(AclOperation operation, ResourcePattern resourcePattern, AuthorizationResult allowed, int error) {
            this.operation = operation;
            this.resourcePattern = resourcePattern;
            this.allowed = allowed;
            this.error = error;
        }

        public AclOperation operation() {
            return operation;
        }

        public ResourcePattern resource() {
            return resourcePattern;
        }

        public AuthorizationResult allowed() {
            return allowed;
        }

        public int errorCode() {
            return error;
        }
    }

    /**
     * Any threads or thread pools can be started in this method. These resources must be closed in the {@link #close()}
     * method.
     */
    void start();

    /**
     * Called on request completion before returning the response to the client. It allows auditing multiple resources
     * in the request, such as multiple topics being created.
     * @param event is the request specific data passed down to the auditor.
 It may be null *if @paramthere requestContextare contains metadata to the request.no specific
     * @param auditInfo is the operation, resource and the outcome of the authorization with theinformation possibleis error
available for the given audited *event type.
     * @param requestContext contains metadata to the request.
     coupled* together.
@param auditAuthInfo is the operation, */
resource and the outcome voidof audit(AuditEvent event, AuthorizableRequestContext requestContext, List<AuditInfo> auditInfo);
}

the authorization with the possible error
     *                  coupled together.
     */
    void audit(AuditEvent event, AuthorizableRequestContext requestContext, List<AuditAuthInfo> auditAuthInfo);
}

package org.apache.kafka.server.auditor.events;

import org.apache.kafka.common.annotation.InterfaceStability;
import org.apache.kafka.server.auditor.AuditEvent;

import java.util.List;

@InterfaceStability.Evolving
public class TopicCreateEventTopicEvent implements AuditEvent {

    public staticenum class CreatedTopicEventType {
         String topicNameCREATE, DELETE, PARTITION_CHANGE, REPLICATION_FACTOR_CHANGE;
     }

    public static class TopicInAudit {
        String topicName;
        int numPartitions;
        int replicationFactor;

        public CreatedTopicTopicInAudit(String topicName, int numPartitions, int replicationFactor) {
            this.topicName = topicName;
            this.numPartitions = numPartitions;
            this.replicationFactor = replicationFactor;
        }

        public String name() {
            return topicName;
        }

        public int numPartitions() {
            return numPartitions;
        }

        public int replicationFactor() {
            return replicationFactor;
        }
    }

    private List<CreatedTopic>List<TopicInAudit> topics;

    publicprivate TopicCreateEvent(List<CreatedTopic> topics) {EventType eventType;

    public TopicEvent(List<TopicInAudit> topics, EventType eventType) {
        this.topics = topics;
        this.eventType = eventType;
    }

    public List<CreatedTopic>List<TopicInAudit> topics() {
        return topics;
    }
}

    public EventType eventType() {
        return eventType;
    }
}