Skip to content

Updates not managed with #1528 #1530

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
masesdevelopers merged 5 commits into from
Apr 10, 2026
Merged

Updates not managed with #1528 #1530

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
c54c56e
Update wrong variable
masesdevelopers Apr 10, 2026
02f4336
Update KNetConsumer
masesdevelopers Apr 10, 2026
3134343
Other fix
masesdevelopers Apr 10, 2026
f22782b
Updates test program
masesdevelopers Apr 10, 2026
63f8200
Update program flow
masesdevelopers Apr 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion src/net/KNet/Developed/Org/Apache/Kafka/Tools/StreamsResetter.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -98,7 +98,7 @@ static bool ResetApplication(bool force, string bootstrapserver, string applicat
string[] parameters = strings.ToArray();

StreamsResetter resetter = new();
var disposable = JVMBridgeCoreDisposable.Create(resetter);
var disposable = MASES.JCOBridge.C2JBridge.JVMBridgeCoreDisposable.Create(resetter);
Java.Lang.String[] args = parameters.ToJVMArray<Java.Lang.String, string>();
try
{
Expand Down
8 changes: 4 additions & 4 deletions src/net/KNet/Specific/Consumer/KNetConsumer.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -196,15 +196,15 @@ static Properties CheckProperties(Properties props, ISerDes keyDeserializer, ISe
public ConsumerRecords<K, V, TJVMK, TJVMV> Poll(TimeSpan timeout)
{
Duration duration = timeout;
var disposable = JVMBridgeCoreDisposable.Create(duration);
var disposable = MASES.JCOBridge.C2JBridge.JVMBridgeCoreDisposable.Create(duration);
try
{
var records = base.Poll(duration);
return new ConsumerRecords<K, V, TJVMK, TJVMV>(records, _keyDeserializer, _valueDeserializer);
}
finally
{
duration?.Dispose();
disposable?.Dispose();
}
}

Expand Down Expand Up @@ -340,15 +340,15 @@ public void Consume(long timeoutMs, Action<ConsumerRecord<K, V, TJVMK, TJVMV>> c
{
Duration duration = TimeSpan.FromMilliseconds(timeoutMs);
if (_consumerCallback == null) throw new ArgumentException("Cannot be used since constructor was called with useJVMCallback set to false.");
var disposable = JVMBridgeCoreDisposable.Create(duration);
var disposable = MASES.JCOBridge.C2JBridge.JVMBridgeCoreDisposable.Create(duration);
try
{
actionCallback = callback;
IExecute("consume", duration);
}
finally
{
duration?.Dispose();
disposable?.Dispose();
actionCallback = null;
}
}
Expand Down
6 changes: 3 additions & 3 deletions tests/net/Benchmarks/KNetBenchmark/ProgramKNet.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -236,7 +236,7 @@ static Stopwatch ConsumeKNet(int testNum, string topicName, int length, int nump

var consumer = KNetConsumer();
var topics = Collections.Singleton((Java.Lang.String)topicName);
var disposable = JVMBridgeCoreDisposable.Create(topics);
var disposable = MASES.JCOBridge.C2JBridge.JVMBridgeCoreDisposable.Create(topics);
try
{
int counter = 0;
Expand Down Expand Up @@ -301,7 +301,7 @@ static Stopwatch ConsumeProduceKNet(string topicName, int length, int numpacket,
var consumer = KNetConsumer();
var producer = KNetProducer();
var topics = Collections.Singleton((Java.Lang.String)topicName);
var disposable = JVMBridgeCoreDisposable.Create(topics);
var disposable = MASES.JCOBridge.C2JBridge.JVMBridgeCoreDisposable.Create(topics);
try
{
int counter = 0;
Expand Down Expand Up @@ -361,7 +361,7 @@ static Stopwatch ConsumeProduceKNet(string topicName, int length, int numpacket,
{
ConsumerRebalanceListener rebalanceListener = null;
var topics = Collections.Singleton((Java.Lang.String)topicName);
var disposable = JVMBridgeCoreDisposable.Create(topics);
var disposable = MASES.JCOBridge.C2JBridge.JVMBridgeCoreDisposable.Create(topics);
try
{
rebalanceListener = new()
Expand Down
8 changes: 4 additions & 4 deletions tests/net/Benchmarks/KNetBenchmark/ProgramKafka.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -291,9 +291,9 @@ static Stopwatch ConsumeKafka(int testNum, string topicName, int length, int num

var consumer = KafkaConsumer();
Java.Time.Duration duration = TimeSpan.FromMinutes(1);
var disposable1 = JVMBridgeCoreDisposable.Create(duration);
var disposable1 = MASES.JCOBridge.C2JBridge.JVMBridgeCoreDisposable.Create(duration);
var topics = Collections.Singleton((Java.Lang.String)topicName);
var disposable2 = JVMBridgeCoreDisposable.Create(topics);
var disposable2 = MASES.JCOBridge.C2JBridge.JVMBridgeCoreDisposable.Create(topics);
try
{
int counter = 0;
Expand Down Expand Up @@ -376,9 +376,9 @@ static Stopwatch ConsumeKafka(int testNum, string topicName, int length, int num
System.Threading.Thread thread = new System.Threading.Thread(() =>
{
Java.Time.Duration duration = TimeSpan.FromSeconds(1);
var disposable1 = JVMBridgeCoreDisposable.Create(duration);
var disposable1 = MASES.JCOBridge.C2JBridge.JVMBridgeCoreDisposable.Create(duration);
var topics = Collections.Singleton((Java.Lang.String)topicName);
var disposable2 = JVMBridgeCoreDisposable.Create(topics);
var disposable2 = MASES.JCOBridge.C2JBridge.JVMBridgeCoreDisposable.Create(topics);
ConsumerRebalanceListener rebalanceListener = null;
try
{
Expand Down
5 changes: 3 additions & 2 deletions tests/net/General/KNetClassicTest/Program.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -90,6 +90,7 @@ private static void Console_CancelKeyPress(object sender, ConsoleCancelEventArgs
static void CreateTopic()
{
NewTopic topic = null;
IDisposable disposable = null;
//Set<NewTopic> coll = null;
try
{
Expand All @@ -98,7 +99,7 @@ static void CreateTopic()
short replicationFactor = 1;

topic = new NewTopic(topicName, partitions, replicationFactor);
var disposable = JVMBridgeCoreDisposable.Create(topic);
disposable = MASES.JCOBridge.C2JBridge.JVMBridgeCoreDisposable.Create(topic);
/**** Direct mode ******
var map = Collections.SingletonMap(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT);
topic.Configs(map);
Expand Down Expand Up @@ -304,7 +305,7 @@ static void ConsumeSomething()
};
}
var topics = Collections.Singleton((Java.Lang.String)topicToUse);
var disposable = JVMBridgeCoreDisposable.Create(topics);
var disposable = MASES.JCOBridge.C2JBridge.JVMBridgeCoreDisposable.Create(topics);
try
{
using (consumer = useSerdes ? new KafkaConsumer<string, string>(props, keyDeserializer, valueDeserializer) : new KafkaConsumer<string, string>(props))
Expand Down
4 changes: 2 additions & 2 deletions tests/net/General/KNetTest/Program.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -402,7 +402,7 @@ static void ConsumeSomething()
Stopwatch watcherTotal = new Stopwatch();
Stopwatch watcher = new Stopwatch();
var topics = Collections.Singleton((Java.Lang.String)topicToUse);
var disposable = JVMBridgeCoreDisposable.Create(topics);
var disposable = MASES.JCOBridge.C2JBridge.JVMBridgeCoreDisposable.Create(topics);
try
{
using (consumer = new KNetConsumer<string, TestType>(props, keyDeserializer, valueDeserializer))
Expand Down Expand Up @@ -631,7 +631,7 @@ static void ConsumeSomethingBuffered()
Stopwatch watcherTotal = new Stopwatch();
Stopwatch watcher = new Stopwatch();
var topics = Collections.Singleton((Java.Lang.String)topicToUse);
var disposable = JVMBridgeCoreDisposable.Create(topics);
var disposable = MASES.JCOBridge.C2JBridge.JVMBridgeCoreDisposable.Create(topics);
try
{
using (consumer = new KNetConsumerValueBuffered<string, TestType>(props, keyDeserializer, valueDeserializer))
Expand Down
Loading