Skip to content

Client support for lightweight consumer #88

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

Open
wants to merge 3 commits into
base: master
Choose a base branch
from
Open

Client support for lightweight consumer #88

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
eebe359
Client support for lightweight consumer
Mrunal98 May 6, 2025
a618996
addressed the review comments.
Mrunal98 May 12, 2025
082b8be
modified seek() for lightweight subscriber.
Mrunal98 May 23, 2025
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
3 changes: 3 additions & 0 deletions clients/src/main/java/org/oracle/okafka/clients/CommonClientConfigs.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,9 @@ public class CommonClientConfigs extends org.apache.kafka.clients.CommonClientCo

public static final String ORACLE_TRANSACTIONAL_PRODUCER ="oracle.transactional.producer";

public static final String ORACLE_CONSUMER_LIGHTWEIGHT = "oracle.consumer.lightweight";
public static final String ORACLE_CONSUMER_LIGHTWEIGHT_DOC = "Creates a light weight subscriber";
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

why do we want that in CommonClientConfigs and ConsumerConfig ?


/*
public static final String METADATA_MAX_AGE_CONFIG = "metadata.max.age.ms";
public static final String METADATA_MAX_AGE_DOC = "The period of time in milliseconds after which we force a refresh of metadata even if we haven't seen any partition leadership changes to proactively discover any new brokers or partitions.";
Expand Down
19 changes: 15 additions & 4 deletions clients/src/main/java/org/oracle/okafka/clients/consumer/ConsumerConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,6 @@
package org.oracle.okafka.clients.consumer;

import org.apache.kafka.clients.ClientDnsLookup;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.common.IsolationLevel;
import org.apache.kafka.common.config.AbstractConfig;
import org.apache.kafka.common.config.ConfigDef;
Expand All @@ -42,6 +41,7 @@
import org.apache.kafka.common.requests.JoinGroupRequest;
import org.apache.kafka.common.serialization.Deserializer;
import org.oracle.okafka.common.config.SslConfigs;
import org.oracle.okafka.clients.CommonClientConfigs;

import java.util.Collections;
import java.util.HashMap;
Expand Down Expand Up @@ -338,15 +338,20 @@ public class ConsumerConfig extends AbstractConfig {
" broker allows for it using `auto.create.topics.enable` broker configuration. This configuration must" +
" be set to `false` when using brokers older than 0.11.0";
public static final boolean DEFAULT_ALLOW_AUTO_CREATE_TOPICS = true;

/**
* <code>security.providers</code>
*/
public static final String SECURITY_PROVIDERS_CONFIG = SecurityConfig.SECURITY_PROVIDERS_CONFIG;
private static final String SECURITY_PROVIDERS_DOC = SecurityConfig.SECURITY_PROVIDERS_DOC;

private static final AtomicInteger CONSUMER_CLIENT_ID_SEQUENCE = new AtomicInteger(1);


/** <code>oracle.consumer.lightweight</code> */
public static final String ORACLE_CONSUMER_LIGHTWEIGHT = "oracle.consumer.lightweight";
public static final String ORACLE_CONSUMER_LIGHTWEIGHT_DOC = "Creates a light weight subscriber";


static {
CONFIG = new ConfigDef().define(BOOTSTRAP_SERVERS_CONFIG,
Type.LIST,
Expand Down Expand Up @@ -606,7 +611,13 @@ public class ConsumerConfig extends AbstractConfig {
.define(org.oracle.okafka.clients.CommonClientConfigs.ORACLE_NET_TNS_ADMIN,
ConfigDef.Type.STRING,
Importance.MEDIUM,
org.oracle.okafka.clients.CommonClientConfigs.ORACLE_NET_TNS_ADMIN_DOC);
org.oracle.okafka.clients.CommonClientConfigs.ORACLE_NET_TNS_ADMIN_DOC)
.define(CommonClientConfigs.ORACLE_CONSUMER_LIGHTWEIGHT,
ConfigDef.Type.BOOLEAN,
false,
Importance.LOW,
CommonClientConfigs.ORACLE_CONSUMER_LIGHTWEIGHT_DOC)
;

}

Expand Down
Loading