Внешняя компонента для работы с Apach Kafka

Использовать компоненту можно на Сервере, так и на Клиенте.

Для использования потребуется Kafka, я запустил в Docker с дефолтными параметрами, а также внешняя компонента (v8Kafka.zip /win64)

Пример использования:

// Producer	
ПутьККомпоненте = "...\v8AddInNative.dll";
	
Попытка
	Результат = ПодключитьВнешнююКомпоненту(ПутьККомпоненте ,"Kafka", ТипВнешнейКомпоненты.Native);
	Компонента = Новый ("AddIn.Kafka.Producer");	
Исключение
	Сообщить(ОписаниеОшибки());
КонецПопытки; 

Компонента.УстановитьПараметр("bootstrap.servers","127.0.0.1:9092");

Компонента.ВызватьОтчетОДоставке = Истина;

Попытка  
	Результат = Компонента.Поставщик();	 
		
	Результат = Компонента.Опубликовать("topic.test", "hello from 1C", "1С", "key1:value1,key2:value2");

	Результат = Компонента.ОжидатьДоставкуСообщений(1000);
		
	ОтчетОДоставке = Компонента.ОтчетОДоставке();
		  
Исключение 
	ОбщегоНазначенияКлиентСервер.СообщитьПользователю(Компонента.ТекстОшибки);
КонецПопытки;

// KafkaConsumer
ПутьККомпоненте = "...\v8AddInNative.dll";
	
Попытка
	Результат = ПодключитьВнешнююКомпоненту(ПутьККомпоненте ,"Kafka", ТипВнешнейКомпоненты.Native, ТипПодключенияВнешнейКомпоненты.НеИзолированно);
	Компонента = Новый("AddIn.Kafka.KafkaConsumer");
Исключение
	Сообщить(ОписаниеОшибки());
КонецПопытки;
		
Попытка  	
		
        Компонента.УстановитьПараметр("bootstrap.servers","127.0.0.1:9092");
	Компонента.УстановитьПараметр("group.id", "group_1");
	Компонента.УстановитьПараметр("enable.auto.commit", "false");
	Компонента.УстановитьПараметр("auto.offset.reset", "earliest");
	Компонента.УстановитьПараметр("enable.partition.eof", "true");
		
	Результат = Компонента.Потребитель();
	Результат = Компонента.Подписаться("topic.test");
		
	Пока Истина Цикл
		Результат = Компонента.Читать(1000);
		
		Если Результат = 0 Тогда 
                        Топик = Компонента.ТопикСообщения();        
			Ключ = Компонента.Ключ();
			Текст = Компонента.Текст(); 
			Заголовки = Компонента.Заголовки();
			Раздел = Компонента.Раздел();
			Смещение = Компонента.Смещение();
			ВремяЧислом = Компонента.ОтметкаВремени();
                        Резельтат = Компонента.Зафиксировать();
		КонецЕсли; 
			
		Если Результат = -191 Тогда // Дошли до конца очереди
			Прервать;
		КонецЕсли;
			
	КонецЦикла;  
			
	Результат = Компонента.Закрыть();
Исключение
КонецПопытки;

// Consumer
ПутьККомпоненте = "...\v8AddInNative.dll";
	
Попытка
	Результат = ПодключитьВнешнююКомпоненту(ПутьККомпоненте ,"Kafka", ТипВнешнейКомпоненты.Native, ТипПодключенияВнешнейКомпоненты.НеИзолированно);
	Компонента = Новый("AddIn.Kafka.Consumer");
Исключение
	Сообщить(ОписаниеОшибки());
КонецПопытки;
		
Попытка  	
		
        Компонента.УстановитьПараметр("bootstrap.servers","127.0.0.1:9092");
	Компонента.УстановитьПараметр("group.id", "group_1");
	Компонента.УстановитьПараметр("auto.offset.reset", "earliest");
        Компонента.УстановитьПараметр("enable.partition.eof", "true");
		
        Результат = Компонента.Потребитель();
	Результат = Компонента.Топик("topic.test");
        Раздел = 0;
        Смещение = "27";
        Результат = Компонента.Старт(Раздел, Смещение); // start(partition, offset)
		
	Пока Истина Цикл
		Результат = Компонента.Читать(Раздел, 1000);
		
		Если Результат = 0 Тогда 
                        Топик = Компонента.Топик();        
			Ключ = Компонента.Ключ();
			Текст = Компонента.Текст(); 
			Заголовки = Компонента.Заголовки();
			Раздел = Компонента.Раздел();
			Смещение = Компонента.Смещение();
			ВремяЧислом = Компонента.ОтметкаВремени();
		КонецЕсли; 
			
		Если Результат = -191 Тогда // Дошли до конца очереди
			Прервать;
		КонецЕсли;
			
	КонецЦикла;  
			
	Результат = Компонента.Стоп(Раздел);
Исключение
КонецПопытки;

Поддерживает:

- сжатие gzip, lz4, zstd

- SSL

Полный список объектов и методов в документации, кратко:

Общие методы

Producer

KafkaConsumer

** компонента предоставлена для тестового использования, будет активна до 23.12.2023