1. Apache Kafka Fundamentals and Integration ============================================

Apache Kafka functions as a distributed streaming platform capable of handling high-throughput data feeds. It operates on three core capabilities: publishing and subscribing to streams of records, storing streams in a fault-tolerant durable manner, and processing streams as they occur.

1.1 Core Concepts

• Topic: A category or feed name to which records are published. Topics are partitioned for scalability.
• Producer: The client that publishes messages to a specific topic.
• Consumer: The client that subscribes to topics and processes the published messages.
• Broker: A server in the Kafka cluster that stores data and serves client requests.

1.2 Native Java Client Implementation

To interact with Kafka programmatically, the kafka-clients library is required.

1.2.1 Message Producer

The producer is responsible for serializing keys and values and sending records to the broker.

import org.apache.kafka.clients.producer.*;
import java.util.Properties;

public class SimpleProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.200.130:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        // Retry configuration
        props.put(ProducerConfig.RETRIES_CONFIG, 10);

        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
            ProducerRecord<String, String> record = new ProducerRecord<>("demo-topic", "key-001", "Hello Kafka World");
            // Synchronous send for demonstration
            producer.send(record, (metadata, exception) -> {
                if (exception != null) {
                    exception.printStackTrace();
                } else {
                    System.out.println("Sent to partition: " + metadata.partition());
                }
            });
        }
    }
}

1.2.2 Message Consumer

Consumers subscribe to topics and poll for new data. They must belong to a consumer group to enable parallel processing and rebalancing.

import org.apache.kafka.clients.consumer.*;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class SimpleConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.200.130:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "demo-consumer-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        // Start reading from earliest offset if no commit exists
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
            consumer.subscribe(Collections.singletonList("demo-topic"));
            
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("Consumed message: Key=%s, Value=%s%n", record.key(), record.value());
                }
                // Manual synchronous commit
                consumer.commitSync();
            }
        }
    }
}

1.3 Spring Boot Integration

Spring Boot abstracts much of the boilerplate configuration required for Kafka via the spring-kafka starter.

1.3.1 Configuration

Define connection properties in application.yml.

server:
  port: 9001
spring:
  kafka:
    bootstrap-servers: 192.168.200.130:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    consumer:
      group-id: spring-boot-group
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

1.3.2 Producer Implementation

Inject KafkaTemplate to send messages.

@RestController
public class MessageController {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @GetMapping("/send")
    public String sendMessage() {
        User user = new User("alice", 25);
        // Convert object to JSON string before sending
        kafkaTemplate.send("spring-topic", JSON.toJSONString(user));
        return "Message Sent";
    }
}

1.3.3 Consumer Implementation

Use the @KafkaListener annotation to handle incoming messages.

@Component
public class MessageListener {

    @KafkaListener(topics = "spring-topic")
    public void onMessage(ConsumerRecord<?, ?> record) {
        Optional<?> message = Optional.ofNullable(record.value());
        message.ifPresent(msg -> {
            System.out.println("Received: " + msg);
            // Optionally deserialize JSON back to Object
        });
    }
}

2. Third-Party Content Security Integration ===========================================

To prevent illegal or inappropriate content from being published, third-party content moderation APIs can be integrated. This example utilizes Alibaba Cloud Content Security services for text and image scanning.

2.1 Text Moderation

Text scanning detects spam, terrorism-related content, and other policy violations. The client sends text content to the API and receives a suggestion (pass, block, or review).

import com.aliyuncs.DefaultAcsClient;
import com.aliyuncs.IAcsClient;
import com.aliyuncs.green.model.v20180509.TextScanRequest;
import com.aliyuncs.http.FormatType;
import com.aliyuncs.http.HttpResponse;
import com.aliyuncs.profile.DefaultProfile;
import org.json.JSONArray;
import org.json.JSONObject;
import java.util.*;

public class TextModerator {
    
    private String accessKeyId = "YOUR_ACCESS_KEY";
    private String accessKeySecret = "YOUR_ACCESS_SECRET";

    public void scanText(String content) throws Exception {
        IClientProfile profile = DefaultProfile.getProfile("cn-shanghai", accessKeyId, accessKeySecret);
        DefaultProfile.addEndpoint("cn-shanghai", "cn-shanghai", "Green", "green.cn-shanghai.aliyuncs.com");
        IAcsClient client = new DefaultAcsClient(profile);

        TextScanRequest request = new TextScanRequest();
        request.setAcceptFormat(FormatType.JSON);
        request.setHttpContentType(FormatType.JSON);
        request.setMethod(com.aliyuncs.http.MethodType.POST);
        request.setEncoding("UTF-8");

        List<Map<String, Object>> tasks = new ArrayList<>();
        Map<String, Object> task = new LinkedHashMap<>();
        task.put("dataId", UUID.randomUUID().toString());
        task.put("content", content);
        tasks.add(task);

        JSONObject data = new JSONObject();
        data.put("scenes", Collections.singletonList("antispam")); // Scene for spam detection
        data.put("tasks", tasks);
        
        request.setHttpContent(data.toString().getBytes("UTF-8"), "UTF-8", FormatType.JSON);
        request.setConnectTimeout(3000);
        request.setReadTimeout(6000);

        HttpResponse response = client.doAction(request);
        if (response.isSuccess()) {
            JSONObject result = new JSONObject(new String(response.getHttpContent(), "UTF-8"));
            // Parse result for 'suggestion': pass, block, review
            System.out.println(result.toString());
        }
    }
}

2.2 Image Moderation

Image moderation involves uploading images (or providing URLs) to detect explicit content, violence, or other violations.

public class ImageModerator {

    private String accessKeyId = "YOUR_ACCESS_KEY";
    private String accessKeySecret = "YOUR_ACCESS_SECRET";

    public void scanImage(byte[] imageData) throws Exception {
        IClientProfile profile = DefaultProfile.getProfile("cn-shanghai", accessKeyId, accessKeySecret);
        DefaultProfile.addEndpoint("cn-shanghai", "cn-shanghai", "Green", "green.cn-shanghai.aliyuncs.com");
        IAcsClient client = new DefaultAcsClient(profile);

        ImageSyncScanRequest request = new ImageSyncScanRequest();
        request.setAcceptFormat(FormatType.JSON);
        request.setMethod(com.aliyuncs.http.MethodType.POST);
        request.setEncoding("UTF-8");

        // Upload image data and get URL (Implementation depends on ClientUploader helper)
        String imageUrl = uploadImage(profile, imageData);

        JSONObject httpBody = new JSONObject();
        httpBody.put("scenes", Arrays.asList("porn", "terrorism")); // Detection scenes
        
        JSONObject task = new JSONObject();
        task.put("dataId", UUID.randomUUID().toString());
        task.put("url", imageUrl);
        httpBody.put("tasks", Collections.singletonList(task));

        request.setHttpContent(httpBody.toString().getBytes("UTF-8"), "UTF-8", FormatType.JSON);
        HttpResponse response = client.doAction(request);
        // Process response logic similar to TextModerator
    }
    
    private String uploadImage(IAcsClient client, byte[] data) {
        // Implementation for uploading to temporary storage provided by SDK helper
        return "http://temp-url-for-scanning.com/image.jpg";
    }
}

3. Sensitive Word Filtering with DFA ====================================

DFA (Deterministic Finite Automaton) provides an efficient algorithm for sensitive word filtering. Unlike database queries, which are slow for large datasets, DFA loads all sensitive words into a tree structure (Map of Maps) in memory, allowing to O(N) search complexity where N is the length of the text to be checked.

3.1 Algorithm Logic

1. Initialization: Build a tree where each character of a sensitive word is a node (Map key). The leaf node contains a flag indicating the end of the word.
2. Detection: Iterate through the input text. If a character matches a root node, move to the next character and deeper into the tree. If an end flag is reached, a match is found.

3.2 Implementation

import java.util.*;

public class DFASensitiveFilter {

    private static Map<String, Object> sensitiveWordMap;

    // Initialize the DFA map with a set of sensitive words
    public static void init(Set<String> words) {
        sensitiveWordMap = new HashMap<>(words.size());
        for (String word : words) {
            Map<String, Object> currentMap = sensitiveWordMap;
            for (int i = 0; i < word.length(); i++) {
                char c = word.charAt(i);
                String key = String.valueOf(c);
                
                Object obj = currentMap.get(key);
                if (obj == null) {
                    Map<String, Object> newMap = new HashMap<>(2);
                    newMap.put("isEnd", false);
                    currentMap.put(key, newMap);
                    currentMap = newMap;
                } else {
                    currentMap = (Map<String, Object>) obj;
                }
                
                if (i == word.length() - 1) {
                    currentMap.put("isEnd", true);
                }
            }
        }
    }

    // Filter sensitive words in text
    public static Set<String> filter(String text) {
        Set<String> foundWords = new HashSet<>();
        for (int i = 0; i < text.length(); i++) {
            int length = checkSensitiveWord(text, i);
            if (length > 0) {
                foundWords.add(text.substring(i, i + length));
                i += length - 1; // Skip matched characters
            }
        }
        return foundWords;
    }

    private static int checkSensitiveWord(String text, int beginIndex) {
        Map<String, Object> currentMap = sensitiveWordMap;
        int matchLength = 0;
        for (int i = beginIndex; i < text.length(); i++) {
            String key = String.valueOf(text.charAt(i));
            currentMap = (Map<String, Object>) currentMap.get(key);
            if (currentMap == null) break;
            
            matchLength++;
            if (Boolean.TRUE.equals(currentMap.get("isEnd"))) {
                return matchLength;
            }
        }
        // If not ended as a full word, reset length
        if (matchLength > 0 && !Boolean.TRUE.equals(currentMap.get("isEnd"))) {
            return 0;
        }
        return matchLength;
    }
}