diff --git a/src/main/java/ru/dmitriymx/minecraft/globalchat/MainPlugin.java b/src/main/java/ru/dmitriymx/minecraft/globalchat/MainPlugin.java index a5822ec..75411c8 100644 --- a/src/main/java/ru/dmitriymx/minecraft/globalchat/MainPlugin.java +++ b/src/main/java/ru/dmitriymx/minecraft/globalchat/MainPlugin.java @@ -21,8 +21,11 @@ public class MainPlugin extends JavaPlugin { } private void initKafkaService() { + ClassLoader originalContext = Thread.currentThread().getContextClassLoader(); + Thread.currentThread().setContextClassLoader(null); //FIXME перенести в конфигурацию service = new KafkaService("127.0.0.1:9092", "global-chat", 1000); + Thread.currentThread().setContextClassLoader(originalContext); mqThread = new Thread(() -> { while (!Thread.currentThread().isInterrupted()) { diff --git a/src/main/java/ru/dmitriymx/minecraft/globalchat/mq/KafkaService.java b/src/main/java/ru/dmitriymx/minecraft/globalchat/mq/KafkaService.java index b89a30b..a04a9ac 100644 --- a/src/main/java/ru/dmitriymx/minecraft/globalchat/mq/KafkaService.java +++ b/src/main/java/ru/dmitriymx/minecraft/globalchat/mq/KafkaService.java @@ -62,19 +62,23 @@ public class KafkaService { } }); + consumer.commitAsync(); + return list; } public void shutdown() { + producer.close(); producer = null; - consumer.unsubscribe(); + + consumer.close(); consumer = null; } private Consumer createConsumer() { final Properties props = new Properties(); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, hosts); - props.put(ConsumerConfig.GROUP_ID_CONFIG, "DefaultGroup"); + props.put(ConsumerConfig.GROUP_ID_CONFIG, "WithoutGroup." + UUID.randomUUID().toString()); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class.getName()); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());