Enhancement 13

.github/workflows/build.yaml

@@ -55,6 +55,7 @@ jobs:
       # Checks-out your repository under $GITHUB_WORKSPACE, so your job can access it
       - uses: actions/checkout@v2
         with:
+          fetch-depth: '1'
           submodules: 'true'
 
       - name: Cache local Maven repository

.github/workflows/pullrequest.yaml

@@ -23,6 +23,7 @@ jobs:
       # Checks-out your repository under $GITHUB_WORKSPACE, so your job can access it                                                                                  
       - uses: actions/checkout@v2
         with:
+          fetch-depth: '1'
           submodules: 'true'
 
       - name: Cache local Maven repository

.github/workflows/release.yaml

@@ -25,6 +25,7 @@ jobs:
       # Checks-out your repository under $GITHUB_WORKSPACE, so your job can access it                                                                                  
       - uses: actions/checkout@v2
         with:
+          fetch-depth: '1'
           submodules: 'true'
 
       - name: Pre compile

src/java/kafkabridge/pom.xml

@@ -9,7 +9,7 @@
     <name>mases.kafkabridge</name>
     <description>Apache Kafka interface bridging implementation</description>
     <url>https://github.com/masesgroup/KafkaBridge</url>
-    <version>1.1.10.0</version>
+    <version>1.1.11.0</version>
 
     <licenses>
         <license>

src/java/kafkabridge/src/main/java/org/mases/kafkabridge/clients/consumer/ConsumerInterceptorImpl.java

@@ -0,0 +1,59 @@
+/*
+ *  MIT License
+ *
+ *  Copyright (c) 2022 MASES s.r.l.
+ *
+ *  Permission is hereby granted, free of charge, to any person obtaining a copy
+ *  of this software and associated documentation files (the "Software"), to deal
+ *  in the Software without restriction, including without limitation the rights
+ *  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ *  copies of the Software, and to permit persons to whom the Software is
+ *  furnished to do so, subject to the following conditions:
+ *
+ *  The above copyright notice and this permission notice shall be included in all
+ *  copies or substantial portions of the Software.
+ *
+ *  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ *  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ *  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ *  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ *  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ *  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ *  SOFTWARE.
+ */
+
+package org.mases.kafkabridge.clients.consumer;
+
+import org.apache.kafka.clients.consumer.ConsumerInterceptor;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.mases.jcobridge.*;
+
+import java.util.Map;
+
+public final class ConsumerInterceptorImpl extends JCListener implements ConsumerInterceptor {
+    public ConsumerInterceptorImpl(String key) throws JCNativeException {
+        super(key);
+    }
+
+    @Override
+    public void configure(Map configs) {
+        raiseEvent("configure", configs);
+    }
+
+    @Override
+    public ConsumerRecords onConsume(ConsumerRecords records) {
+        raiseEvent("onConsume", records);
+        Object retVal = getReturnData();
+        return (ConsumerRecords) retVal;
+    }
+
+    @Override
+    public void onCommit(Map offsets) {
+        raiseEvent("onCommit", offsets);
+    }
+
+    @Override
+    public void close() {
+        raiseEvent("close");
+    }
+}

src/java/kafkabridge/src/test/java/org/mases/kafkabridge/clients/consumer/ConsumerInterceptorImplTest.java

@@ -0,0 +1,18 @@
+package org.mases.kafkabridge.clients.consumer;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Unit test for simple App.
+ */
+public class ConsumerInterceptorImplTest {
+    /**
+     * Rigorous Test :-)
+     */
+    @Test
+    public void shouldAnswerWithTrue() {
+        assertTrue(true);
+    }
+}

src/net/KafkaBridge/ClientSide/BridgedClasses/Clients/Consumer/ConsumerInterceptor.cs

@@ -0,0 +1,157 @@
+/*
+*  Copyright 2022 MASES s.r.l.
+*
+*  Licensed 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.
+*
+*  Refer to LICENSE for more information.
+*/
+
+using MASES.JCOBridge.C2JBridge;
+using MASES.KafkaBridge.Common;
+using Java.Util;
+using System;
+
+namespace MASES.KafkaBridge.Clients.Consumer
+{
+    /// <summary>
+    /// Listener for Kafka ConsumerInterceptor. Extends <see cref="IJVMBridgeBase"/>
+    /// </summary>
+    public interface IConsumerInterceptor<K, V> : IJVMBridgeBase
+    {
+        /// <summary>
+        /// Configure this class with the given key-value pairs
+        /// </summary>
+        /// <param name="configs">The configuration <see cref="Map"/></param>
+        void Configure(Map<string, Java.Lang.Object> configs);
+        /// <summary>
+        /// This is called just before the records are returned by <see cref="KafkaConsumer{K,V}.Poll(Java.Time.Duration)"/>
+        /// </summary>
+        /// <param name="records">records records to be consumed by the client or records returned by the previous interceptors in the list</param>
+        /// <returns>records that are either modified by the interceptor or same as records passed to this method</returns>
+        ConsumerRecords<K, V> OnConsume(ConsumerRecords<K, V> records);
+        /// <summary>
+        /// This is called when offsets get committed.
+        /// </summary>
+        /// <param name="offsets">A <see cref="Map"/> of offsets by partition with associated metadata</param>
+        void OnCommit(Map<TopicPartition, OffsetAndMetadata> offsets);
+        /// <summary>
+        /// This is called when interceptor is closed
+        /// </summary>
+        void Close();
+    }
+
+    /// <summary>
+    /// Listener for Kafka ConsumerRebalanceListener. Extends <see cref="JVMBridgeListener"/>, implements <see cref="IConsumerInterceptor{K, V}"/>
+    /// </summary>
+    /// <remarks>Remember to Dispose the object otherwise there is a resource leak, the object contains a reference to the the corresponding JVM object</remarks>
+    public class ConsumerInterceptor<K, V> : JVMBridgeListener, IConsumerInterceptor<K, V>
+    {
+        /// <inheritdoc cref="JVMBridgeListener.ClassName"/>
+        public sealed override string ClassName => "org.mases.kafkabridge.clients.consumer.ConsumerInterceptorImpl";
+
+        readonly Action<Map<string, Java.Lang.Object>> configureFunction = null;
+        readonly Func<ConsumerRecords<K, V>, ConsumerRecords<K, V>> onConsumeFunction = null;
+        readonly Action<Map<TopicPartition, OffsetAndMetadata>> onCommitFunction = null;
+        readonly Action closeFunction = null;
+        /// <summary>
+        /// The <see cref="Action{Map{string, Java.Lang.Object}}"/> to be executed on Configure
+        /// </summary>
+        public virtual Action<Map<string, Java.Lang.Object>> OnConfigure { get { return configureFunction; } }
+        /// <summary>
+        /// The <see cref="Func{ConsumerRecords{K, V}, ConsumerRecords{K, V}}"/> to be executed on OnConsume
+        /// </summary>
+        public virtual Func<ConsumerRecords<K, V>, ConsumerRecords<K, V>> OnOnConsume { get { return onConsumeFunction; } }
+        /// <summary>
+        /// The <see cref="Action{Map{TopicPartition, OffsetAndMetadata}}"/> to be executed on OnCommit
+        /// </summary>
+        public virtual Action<Map<TopicPartition, OffsetAndMetadata>> OnOnCommit { get { return onCommitFunction; } }
+        /// <summary>
+        /// The <see cref="Action"/> to be executed on Close
+        /// </summary>
+        public virtual Action OnClose { get { return closeFunction; } }
+        /// <summary>
+        /// Initialize a new instance of <see cref="ConsumerInterceptor"/>
+        /// </summary>
+        /// <param name="onConsume">The <see cref="Action{Map{string, Java.Lang.Object}}"/> to be executed on Configure</param>
+        /// <param name="onConsume">The <see cref="Func{ConsumerRecords{K, V}, ConsumerRecords{K, V}}"/> to be executed on OnConsume</param>
+        /// <param name="onCommit">The <see cref="Action{Map{TopicPartition, OffsetAndMetadata}}"/> to be executed on OnCommit</param>
+        /// <param name="onClose">The <see cref="Action"/> to be executed on Close</param>
+        /// <param name="attachEventHandler">Set to false to disable attach of <see cref="EventHandler"/> and set an own one</param>
+        public ConsumerInterceptor(Action<Map<string, Java.Lang.Object>> configure = null,
+                                   Func<ConsumerRecords<K, V>, ConsumerRecords<K, V>> onConsume = null,
+                                   Action<Map<TopicPartition, OffsetAndMetadata>> onCommit = null,
+                                   Action onClose = null,
+                                   bool attachEventHandler = true)
+        {
+            if (configure != null) configureFunction = configure;
+            else configureFunction = Configure;
+            if (onConsume != null) onConsumeFunction = onConsume;
+            else onConsumeFunction = OnConsume;
+            if (onCommit != null) onCommitFunction = onCommit;
+            else onCommitFunction = OnCommit;
+            if (onClose != null) closeFunction = onClose;
+            else closeFunction = Close;
+
+            if (attachEventHandler)
+            {
+                AddEventHandler("configure", new EventHandler<CLRListenerEventArgs<CLREventData<Map<string, Java.Lang.Object>>>>(EventHandlerConfigure));
+                AddEventHandler("onConsume", new EventHandler<CLRListenerEventArgs<CLREventData<ConsumerRecords<K, V>>>>(EventHandlerOnConsume));
+                AddEventHandler("onCommit", new EventHandler<CLRListenerEventArgs<CLREventData<Map<TopicPartition, OffsetAndMetadata>>>>(EventHandlerOnCommit));
+                AddEventHandler("close", new EventHandler<CLRListenerEventArgs<CLREventData>>(EventHandlerClose));
+            }
+        }
+
+        void EventHandlerConfigure(object sender, CLRListenerEventArgs<CLREventData<Map<string, Java.Lang.Object>>> data)
+        {
+            OnConfigure(data.EventData.TypedEventData);
+        }
+
+        void EventHandlerOnConsume(object sender, CLRListenerEventArgs<CLREventData<ConsumerRecords<K, V>>> data)
+        {
+            var result = OnOnConsume(data.EventData.TypedEventData);
+            data.SetReturnValue(result);
+        }
+
+        void EventHandlerOnCommit(object sender, CLRListenerEventArgs<CLREventData<Map<TopicPartition, OffsetAndMetadata>>> data)
+        {
+            OnOnCommit(data.EventData.TypedEventData);
+        }
+
+        void EventHandlerClose(object sender, CLRListenerEventArgs<CLREventData> data)
+        {
+            OnClose();
+        }
+
+        /// <inheritdoc cref="IConsumerInterceptor{K, V}.Configure"/>
+        public virtual void Configure(Map<string, Java.Lang.Object> configs)
+        {
+
+        }
+
+        /// <inheritdoc cref="IConsumerInterceptor{K, V}.OnConsume"/>
+        public virtual ConsumerRecords<K, V> OnConsume(ConsumerRecords<K, V> records)
+        {
+            return default;
+        }
+
+        /// <inheritdoc cref="IConsumerInterceptor{K, V}.OnCommit"/>
+        public virtual void OnCommit(Map<TopicPartition, OffsetAndMetadata> offsets)
+        {
+        }
+
+        /// <inheritdoc cref="IConsumerInterceptor{K, V}.Close"/>
+        public virtual void Close()
+        {
+        }
+    }
+}

src/net/KafkaBridge/ClientSide/BridgedClasses/Clients/Consumer/ConsumerRebalanceListener.cs

@@ -64,15 +64,18 @@ public class ConsumerRebalanceListener : JVMBridgeListener, IConsumerRebalanceLi
         /// </summary>
         /// <param name="revoked">The <see cref="Action{Collection{TopicPartition}}"/> to be executed on revoked partitions</param>
         /// <param name="assigned">The <see cref="Action{Collection{TopicPartition}}"/> to be executed on assigned partitions</param>
-        public ConsumerRebalanceListener(Action<Collection<TopicPartition>> revoked = null, Action<Collection<TopicPartition>> assigned = null)
+        /// <param name="attachEventHandler">Set to false to disable attach of <see cref="EventHandler"/> and set an own one</param>
+        public ConsumerRebalanceListener(Action<Collection<TopicPartition>> revoked = null, Action<Collection<TopicPartition>> assigned = null, bool attachEventHandler = true)
         {
             if (revoked != null) revokedFunction = revoked;
             else revokedFunction = OnPartitionsRevoked;
             if (assigned != null) assignedFunction = assigned;
             else assignedFunction = OnPartitionsAssigned;
-
-            AddEventHandler("onPartitionsRevoked", new EventHandler<CLRListenerEventArgs<CLREventData<Collection<TopicPartition>>>>(EventHandlerRevoked));
-            AddEventHandler("onPartitionsAssigned", new EventHandler<CLRListenerEventArgs<CLREventData<Collection<TopicPartition>>>>(EventHandlerAssigned));
+            if (attachEventHandler)
+            {
+                AddEventHandler("onPartitionsRevoked", new EventHandler<CLRListenerEventArgs<CLREventData<Collection<TopicPartition>>>>(EventHandlerRevoked));
+                AddEventHandler("onPartitionsAssigned", new EventHandler<CLRListenerEventArgs<CLREventData<Collection<TopicPartition>>>>(EventHandlerAssigned));
+            }
         }
 
         void EventHandlerRevoked(object sender, CLRListenerEventArgs<CLREventData<Collection<TopicPartition>>> data)

src/net/KafkaBridge/ClientSide/BridgedClasses/Clients/Consumer/ConsumerRecords.cs

@@ -16,15 +16,30 @@
 *  Refer to LICENSE for more information.
 */
 
+using Java.Lang;
+using Java.Util;
+using MASES.KafkaBridge.Common;
+
 namespace MASES.KafkaBridge.Clients.Consumer
 {
     public class ConsumerRecords<K, V> : JCOBridge.C2JBridge.JVMBridgeBaseEnumerable<ConsumerRecords<K, V>, ConsumerRecord<K, V>>
     {
         public override string ClassName => "org.apache.kafka.clients.consumer.ConsumerRecords";
-    }
 
-    public class ConsumerRecords : ConsumerRecords<object, object>
-    {
+        public List<ConsumerRecord<K, V>> Records(TopicPartition partition)
+        {
+            return IExecute<List<ConsumerRecord<K, V>>>("records", partition);
+        }
+
+        public Iterable<ConsumerRecord<K, V>> Records(String topic)
+        {
+            return IExecute<Iterable<ConsumerRecord<K, V>>>("records", topic);
+        }
+
+        public Set<TopicPartition> partitions => IExecute<Set<TopicPartition>>("partitions");
+
+        public Iterator<ConsumerRecord<K, V>> Iterator => IExecute<Iterator<ConsumerRecord<K, V>>>("iterator");
 
+        public int Count => IExecute<int>("count");
     }
 }

src/net/KafkaBridge/ClientSide/BridgedClasses/Clients/Consumer/KafkaConsumer.cs

@@ -24,7 +24,7 @@
 
 namespace MASES.KafkaBridge.Clients.Consumer
 {
-    public class KafkaConsumer<K, V> : JCOBridge.C2JBridge.JVMBridgeBase<KafkaConsumer<K, V>>, IConsumer<K, V>
+    public class KafkaConsumer : JCOBridge.C2JBridge.JVMBridgeBase<KafkaConsumer>
     {
         public override bool IsCloseable => true;
 
@@ -34,6 +34,18 @@ public KafkaConsumer()
         {
         }
 
+        public KafkaConsumer(params object[] args)
+            : base(args)
+        {
+        }
+    }
+
+    public class KafkaConsumer<K, V> : KafkaConsumer, IConsumer<K, V>
+    {
+        public KafkaConsumer()
+        {
+        }
+
         public KafkaConsumer(Properties props)
             : base(props)
         {
@@ -257,9 +269,4 @@ public void Wakeup()
             IExecute("wakeup");
         }
     }
-
-    public class KafkaConsumer : KafkaConsumer<object, object>
-    {
-        public KafkaConsumer(Properties props) : base(props) { }
-    }
 }