Skip to content

Removing offline schema access #38

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 2 commits into
base: master
Choose a base branch
from
Open

Removing offline schema access #38

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
2bea3e3
Removing offline schema access
ruchirvaninasdaq Jan 6, 2023
0174db3
Use config for timeout
ruchirvaninasdaq Jan 13, 2023
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 ncdssdk/src/main/python/ncdsclient/consumer/NasdaqKafkaAvroConsumer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -93,7 +93,7 @@ def get_kafka_consumer(self, stream_name, timestamp=None):
if auto_offset_cfg == "earliest" or auto_offset_cfg == "smallest" or auto_offset_cfg == "beginning":
self.logger.debug(
f"Auto offset reset config set to: {auto_offset_cfg}")
return SeekToMidnight.seek_to_midnight_at_past_day(kafka_consumer, topic_partition, 0)
return SeekToMidnight.seek_to_midnight_at_past_day(kafka_consumer, topic_partition, 0, self.kafka_props.get(self.kafka_config_loader.TIMEOUT))

else:
return kafka_consumer
Expand Down
9 changes: 2 additions & 7 deletions ncdssdk/src/main/python/ncdsclient/internal/ReadSchemaTopic.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,6 @@
from ncdssdk.src.main.python.ncdsclient.internal.utils.AuthenticationConfigLoader import AuthenticationConfigLoader
from ncdssdk.src.main.python.ncdsclient.internal.utils.KafkaConfigLoader import KafkaConfigLoader
import ncdssdk.src.main.resources as sysresources
import ncdssdk.src.main.resources.schemas as schemas
from ncdssdk.src.main.python.ncdsclient.internal.KafkaAvroConsumer import KafkaAvroConsumer
from confluent_kafka import TopicPartition
from confluent_kafka import OFFSET_BEGINNING
Expand Down Expand Up @@ -63,10 +62,6 @@ def read_schema(self, topic):
message_schema = avro.schema.parse(latest_record_val['schema'])
schema_consumer.close()

if not message_schema:
print("WARNING: Using the Old Schema! It might not be the latest schema.")
message_schema = self.internal_schema(topic)

self.logger.debug("Returning message schema in read_schema")

return message_schema
Expand Down Expand Up @@ -106,7 +101,7 @@ def get_consumer(self, client_id):
self.kafka_props = KafkaConfigLoader.load_test_config()

self.kafka_props[self.kafka_config_loader.AUTO_OFFSET_RESET_CONFIG] = 'earliest'
self.kafka_props[self.kafka_config_loader.GROUP_ID_CONFIG] = f'{client_id}1'
self.kafka_props[self.kafka_config_loader.GROUP_ID_CONFIG] = client_id

kafka_avro_consumer = KafkaAvroConsumer(
self.kafka_props, ctrl_msg_schema)
Expand All @@ -116,7 +111,7 @@ def get_consumer(self, client_id):

kafka_avro_consumer.assign([topic_partition])

return SeekToMidnight.seek_to_midnight_at_past_day(kafka_avro_consumer, topic_partition, 7)
return SeekToMidnight.seek_to_midnight_at_past_day(kafka_avro_consumer, topic_partition, 6, self.kafka_props[self.kafka_config_loader.TIMEOUT])

def internal_schema(self, topic):
try:
Expand Down
4 changes: 2 additions & 2 deletions ncdssdk/src/main/python/ncdsclient/internal/utils/SeekToMidnight.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,14 +6,14 @@
logger = logging.getLogger(__name__)


def seek_to_midnight_at_past_day(kafka_avro_consumer, topic_partition, num_days_ago=0):
def seek_to_midnight_at_past_day(kafka_avro_consumer, topic_partition, num_days_ago=0, timeout=10):
topic_partition.offset = get_timestamp_at_midnight(num_days_ago)
logger.debug(
f"Num days ago: {num_days_ago}. Setting partition offset to timestamp: {topic_partition.offset}")
try:
logger.debug(f"topic partition: {topic_partition}")
offsets_for_times = kafka_avro_consumer.offsets_for_times(
[topic_partition], timeout=5)
[topic_partition], timeout=timeout)
except Exception as e:
logger.exception(e)
sys.exit(0)
Expand Down
4 changes: 0 additions & 4 deletions ncdssdk/src/main/resources/consumer-properties.json
View file
Open in desktop

This file was deleted.

Loading